Async iteration & streams.

export type PushFeed = Readonly<{
	stream: ReadableStream<Update>;
	publish(update: Update): void;
	close(): void;
}>;

// Push: the source owns the controller and publishes before a reader asks.
export function createPushFeed(stats: FeedStats = newStats()): PushFeed {
	let controller: ReadableStreamDefaultController<Update> | undefined;
	let active = true;
	const stream = new ReadableStream<Update>({
		start(next) {
			controller = next;
		},
		cancel() {
			active = false;
			stats.cancelled = true;
			stats.closed = true;
		}
	});

	return {
		stream,
		publish(update) {
			if (!active) return;
			stats.produced += 1;
			controller?.enqueue({ ...update });
		},
		close() {
			if (!active) return;
			active = false;
			stats.closed = true;
			controller?.close();
		}
	};
}

export async function consumePull(
	rows: readonly Update[],
	limit: number,
	stats = newStats()
): Promise<Update[]> {
	const result: Update[] = [];
	const iterator = pullUpdates(rows, stats);
	try {
		for await (const row of iterator) {
			result.push(row);
			stats.delivered += 1;
			if (result.length === limit) break;
		}
	} finally {
		await iterator.return?.(undefined);
	}
	return result;
}

export async function consumePush(
	stream: ReadableStream<Update>,
	limit: number,
	stats: FeedStats
): Promise<Update[]> {
	const result: Update[] = [];
	const reader = stream.getReader();
	try {
		while (result.length < limit) {
			const next = await reader.read();
			if (next.done) break;
			result.push(next.value);
			stats.delivered += 1;
		}
		if (result.length === limit && !stats.closed) await reader.cancel('consumer stopped');
	} finally {
		reader.releaseLock();
	}
	return result;
}

// Channel range is push delivery: the producer sends, and the receiver ranges.
func streamUpdates(ctx context.Context, rows []Update) <-chan Update {
	out := make(chan Update)
	go func() {
		defer close(out)
		for _, row := range rows {
			select {
			case out <- row:
			case <-ctx.Done():
				return
			}
		}
	}()
	return out
}

func collectPull(rows []Update, limit int) []Update {
	feed := NewPullFeed(rows)
	defer feed.Close()
	result := make([]Update, 0, limit)
	for len(result) < limit {
		row, ok := feed.Next()
		if !ok {
			break
		}
		result = append(result, row)
	}
	return result
}

// collectChannel owns the context it hands the producer. Breaking the range
// leaves no receiver, so the deferred cancel is the producer's only way out.
func collectChannel(ctx context.Context, rows []Update, limit int) []Update {
	ctx, cancel := context.WithCancel(ctx)
	defer cancel()
	result := make([]Update, 0, limit)
	for row := range streamUpdates(ctx, rows) {
		result = append(result, row)
		if len(result) == limit {
			break
		}
	}
	return result
}