feat: implement select statement compliance

Signed-off-by: Christian Stewart <christian@aperture.us>

Author: paralin

builtin/builtin.ts

@@ -104,6 +104,24 @@ export const append = <T>(slice: Array<T>, ...elements: T[]): Array<T> => {
     slice.push(...elements);
     return slice;
 };
+
+/**
+ * Represents the result of a channel receive operation with 'ok' value
+ */
+export interface ChannelReceiveResult<T> {
+    value: T; // Should be T | ZeroValue<T>
+    ok: boolean;
+}
+
+/**
+ * Represents a result from a select operation
+ */
+export interface SelectResult<T> {
+    value: T; // Should be T | ZeroValue<T>
+    ok: boolean;
+    id: number;
+}
+
 /**
  * Represents a Go channel in TypeScript.
  * Supports asynchronous sending and receiving of values.
@@ -119,14 +137,53 @@ export interface Channel<T> {
     /**
      * Receives a value from the channel.
      * Returns a promise that resolves with the received value.
+     * If the channel is closed, it throws an error.
      */
     receive(): Promise<T>;
 
+    /**
+     * Receives a value from the channel along with a boolean indicating
+     * whether the channel is still open.
+     * Returns a promise that resolves with {value, ok}.
+     * - If channel is open and has data: {value: <data>, ok: true}
+     * - If channel is closed and empty: {value: <zero value>, ok: false}
+     * - If channel is closed but has remaining buffered data: {value: <data>, ok: true}
+     */
+    receiveWithOk(): Promise<ChannelReceiveResult<T>>;
+
     /**
      * Closes the channel.
      * No more values can be sent to a closed channel.
+     * Receive operations on a closed channel return the zero value and ok=false.
      */
     close(): void;
+
+    /**
+     * Used in select statements to create a receive operation promise.
+     * @param id An identifier for this case in the select statement
+     * @returns Promise that resolves when this case is selected
+     */
+    selectReceive(id: number): Promise<SelectResult<T>>;
+
+    /**
+     * Used in select statements to create a send operation promise.
+     * @param value The value to send
+     * @param id An identifier for this case in the select statement
+     * @returns Promise that resolves when this case is selected
+     */
+    selectSend(value: T, id: number): Promise<SelectResult<boolean>>;
+
+    /**
+     * Checks if the channel has data ready to be received without blocking.
+     * Used for non-blocking select operations.
+     */
+    canReceiveNonBlocking(): boolean;
+
+    /**
+     * Checks if the channel can accept a send operation without blocking.
+     * Used for non-blocking select operations.
+     */
+    canSendNonBlocking(): boolean;
 }
 
 // A simple implementation of buffered channels
@@ -136,9 +193,12 @@ class BufferedChannel<T> implements Channel<T> {
     private capacity: number;
     private senders: Array<(value: boolean) => void> = []; // Resolvers for blocked senders
     private receivers: Array<(value: T) => void> = []; // Resolvers for blocked receivers
+    private receiversWithOk: Array<(result: ChannelReceiveResult<T>) => void> = []; // For receive with ok
+    private zeroValue: T; // Store the zero value for the element type
 
-    constructor(capacity: number) {
+    constructor(capacity: number, zeroValue: T) {
         this.capacity = capacity;
+        this.zeroValue = zeroValue;
     }
 
     async send(value: T): Promise<void> {
@@ -153,6 +213,13 @@ class BufferedChannel<T> implements Channel<T> {
             return;
         }
 
+        // If there are waiting receivers with ok, directly pass the value and ok=true
+        if (this.receiversWithOk.length > 0) {
+            const receiver = this.receiversWithOk.shift()!;
+            receiver({ value, ok: true });
+            return;
+        }
+
         // If buffer is not full, add to buffer
         if (this.buffer.length < this.capacity) {
             this.buffer.push(value);
@@ -186,7 +253,7 @@ class BufferedChannel<T> implements Channel<T> {
             return value;
         }
 
-        // If channel is closed and buffer is empty, throw error
+        // If channel is closed and buffer is empty, throw error (receive without ok)
         if (this.closed) {
             throw new Error("receive on closed channel");
         }
@@ -197,6 +264,96 @@ class BufferedChannel<T> implements Channel<T> {
         });
     }
 
+    async receiveWithOk(): Promise<ChannelReceiveResult<T>> {
+        // If buffer has values, return from buffer with ok=true
+        if (this.buffer.length > 0) {
+            const value = this.buffer.shift()!;
+
+            // If there are waiting senders, unblock one
+            if (this.senders.length > 0) {
+                const sender = this.senders.shift()!;
+                sender(true); // Unblock with success
+            }
+
+            return { value, ok: true };
+        }
+
+        // If channel is closed and buffer is empty, return zero value with ok=false
+        if (this.closed) {
+            return { value: this.zeroValue, ok: false }; // Return zeroValue
+        }
+
+        // Buffer is empty, block the receiver with ok
+        return new Promise<ChannelReceiveResult<T>>((resolve) => {
+            this.receiversWithOk.push(resolve);
+        });
+    }
+
+    async selectReceive(id: number): Promise<SelectResult<T>> {
+        // If buffer has values, return from buffer with ok=true
+        if (this.buffer.length > 0) {
+            const value = this.buffer.shift()!;
+
+            // If there are waiting senders, unblock one
+            if (this.senders.length > 0) {
+                const sender = this.senders.shift()!;
+                sender(true); // Unblock with success
+            }
+
+            return { value, ok: true, id };
+        }
+
+        // If channel is closed and buffer is empty, return zero value with ok=false
+        if (this.closed) {
+            return { value: this.zeroValue, ok: false, id }; // Return zeroValue
+        }
+
+        // Buffer is empty, return a promise that will be resolved when a value is available
+        return new Promise<SelectResult<T>>((resolve) => {
+            this.receiversWithOk.push((result) => {
+                resolve({ ...result, id });
+            });
+        });
+    }
+
+    async selectSend(value: T, id: number): Promise<SelectResult<boolean>> {
+        if (this.closed) {
+            return { value: false, ok: false, id };
+        }
+
+        // If there are waiting receivers, directly pass the value
+        if (this.receivers.length > 0) {
+            const receiver = this.receivers.shift()!;
+            receiver(value);
+            return { value: true, ok: true, id };
+        }
+
+        // If there are waiting receivers with ok, directly pass the value and ok=true
+        if (this.receiversWithOk.length > 0) {
+            const receiver = this.receiversWithOk.shift()!;
+            receiver({ value, ok: true });
+            return { value: true, ok: true, id };
+        }
+
+        // If buffer is not full, add to buffer
+        if (this.buffer.length < this.capacity) {
+            this.buffer.push(value);
+            return { value: true, ok: true, id };
+        }
+
+        // Buffer is full, return a promise that will be resolved when buffer space is available
+        return new Promise<SelectResult<boolean>>((resolve, reject) => {
+            this.senders.push((success: boolean) => {
+                if (success) {
+                    this.buffer.push(value);
+                    resolve({ value: true, ok: true, id });
+                } else {
+                    resolve({ value: false, ok: false, id });
+                }
+            });
+        });
+    }
+
     close(): void {
         if (this.closed) {
             throw new Error("close of closed channel");
@@ -210,19 +367,141 @@ class BufferedChannel<T> implements Channel<T> {
         }
         this.senders = [];
 
-        // Unblock all waiting receivers with undefined
+        // Unblock all waiting receivers with the zero value
         for (const receiver of this.receivers) {
-            receiver(undefined as any);
+             // Note: receive() without ok throws on closed channel, this unblocking should not happen in correct Go logic
+             // but for safety, we'll resolve with zero value if it somehow does.
+            receiver(this.zeroValue);
         }
         this.receivers = [];
+
+        // Unblock all waiting receivers with ok=false and zero value
+        for (const receiver of this.receiversWithOk) {
+            receiver({ value: this.zeroValue, ok: false });
+        }
+        this.receiversWithOk = [];
+    }
+
+    /**
+     * Checks if the channel has data ready to be received without blocking.
+     * Used for non-blocking select operations.
+     */
+    canReceiveNonBlocking(): boolean {
+        return this.buffer.length > 0 || this.closed;
+    }
+
+    /**
+     * Checks if the channel can accept a send operation without blocking.
+     * Used for non-blocking select operations.
+     */
+    canSendNonBlocking(): boolean {
+        return !this.closed && this.buffer.length < this.capacity;
+    }
+}
+
+/**
+ * Represents a case in a select statement.
+ */
+export interface SelectCase<T> {
+    id: number;
+    isSend: boolean; // true for send, false for receive
+    channel: Channel<any>;
+    value?: any; // Value to send for send cases
+    // Optional handlers for when this case is selected
+    onSelected?: (result: SelectResult<T>) => void | Promise<void>;
+}
+
+
+/**
+ * Helper for 'select' statements. Takes an array of select cases
+ * and resolves when one of them completes, following Go's select rules.
+ *
+ * @param cases Array of SelectCase objects
+ * @param hasDefault Whether there is a default case
+ * @returns A promise that resolves with the result of the selected case
+ */
+export async function selectStatement<T>(
+    cases: SelectCase<T>[],
+    hasDefault: boolean = false
+): Promise<void> { // Changed return type to void as onSelected handles the case body
+    if (cases.length === 0 && !hasDefault) {
+         // Go spec: If there are no cases, the select statement blocks forever.
+         // Emulate blocking forever with a promise that never resolves.
+        return new Promise<void>(() => {}); // Promise never resolves
+    }
+
+    // 1. Check for ready (non-blocking) operations
+    const readyCases: SelectCase<T>[] = [];
+    for (const caseObj of cases) {
+        if (caseObj.id === -1) { // Skip default case in this check
+            continue
+        }
+        if (caseObj.isSend) {
+            if (caseObj.channel.canSendNonBlocking()) {
+                readyCases.push(caseObj);
+            }
+        } else {
+            if (caseObj.channel.canReceiveNonBlocking()) {
+                readyCases.push(caseObj);
+            }
+        }
+    }
+
+    if (readyCases.length > 0) {
+        // If one or more cases are ready, choose one pseudo-randomly
+        const selectedCase = readyCases[Math.floor(Math.random() * readyCases.length)];
+
+        // Execute the selected operation and its onSelected handler
+        if (selectedCase.isSend) {
+            const result = await selectedCase.channel.selectSend(selectedCase.value, selectedCase.id);
+             if (selectedCase.onSelected) {
+                await selectedCase.onSelected(result as SelectResult<T>); // Await the handler
+            }
+        } else {
+            const result = await selectedCase.channel.selectReceive(selectedCase.id);
+             if (selectedCase.onSelected) {
+                await selectedCase.onSelected(result); // Await the handler
+            }
+        }
+        return; // Return after executing a ready case
+    }
+
+    // 2. If no operations are ready and there's a default case, select default
+    if (hasDefault) {
+        // Find the default case (it will have id -1)
+        const defaultCase = cases.find(c => c.id === -1);
+        if (defaultCase && defaultCase.onSelected) {
+             // Execute the onSelected handler for the default case
+             await defaultCase.onSelected({ value: undefined, ok: false, id: -1 } as SelectResult<T>); // Await the handler
+        }
+        return; // Return after executing the default case
+    }
+
+    // 3. If no operations are ready and no default case, block until one is ready
+    // Use Promise.race on the blocking promises
+    const blockingPromises = cases.filter(c => c.id !== -1).map(caseObj => { // Exclude default case
+        if (caseObj.isSend) {
+            return caseObj.channel.selectSend(caseObj.value, caseObj.id);
+        } else {
+            return caseObj.channel.selectReceive(caseObj.id);
+        }
+    });
+
+    const result = await Promise.race(blockingPromises);
+     // Execute onSelected handler for the selected case
+    const selectedCase = cases.find(c => c.id === result.id);
+    if (selectedCase && selectedCase.onSelected) {
+        await selectedCase.onSelected(result); // Await the handler
     }
+    // No explicit return needed here, as the function will implicitly return after the await
 }
 
 /**
- * Creates a new channel with the specified buffer size.
+ * Creates a new channel with the specified buffer size and zero value.
  * @param bufferSize The size of the channel buffer. If 0, creates an unbuffered channel.
+ * @param zeroValue The zero value for the channel's element type.
  * @returns A new channel instance.
  */
-export const makeChannel = <T>(bufferSize: number): Channel<T> => {
-    return new BufferedChannel<T>(bufferSize);
+export const makeChannel = <T>(bufferSize: number, zeroValue: T): Channel<T> => {
+    return new BufferedChannel<T>(bufferSize, zeroValue);
 };