/**
 * TopicDO — topic-based pub/sub fan-out.
 *
 * One DO instance per topic key (e.g. "topic:chat:thread:t1", "topic:live:deal:d1").
 * Maintains a subscriber list of UserSessionDO stub names.
 * Coalesces high-frequency events (deal.viewer_count_tick, chat.typing, etc.)
 * via alarm-based flush. Low-frequency events are fanned out immediately.
 *
 * GC: an idle alarm fires after GC_IDLE_MS with no subscribers — deletes all storage.
 */
import { DurableObject } from 'cloudflare:workers';
import type { MultidealEnv } from '../lib/env.js';

type Envelope = {
  ch: string;
  kind: string;
  tier: string;
  event?: string;
  payload?: unknown;
  ts: number;
  msgId?: string;
};

/** Events that must be coalesced (batched) rather than fanned-out immediately. */
const COALESCED = new Set(['deal.viewer_count_tick', 'groupdeal.slot_filled', 'chat.typing']);

const FAN_OUT_BATCH = 100;
const GC_IDLE_MS = 5 * 60_000;

export class TopicDO extends DurableObject<MultidealEnv> {
  // ─── Subscriber management ───────────────────────────────────────────────

  async addSubscriber(userStubName: string): Promise<void> {
    const subs = new Set((await this.ctx.storage.get<string[]>('subs')) ?? []);
    subs.add(userStubName);
    await this.ctx.storage.put('subs', [...subs]);
    // Cancel any pending GC alarm if it was a gc type.
    const reason = await this.ctx.storage.get<'flush' | 'gc'>('alarmReason');
    if (reason === 'gc') {
      await this.ctx.storage.deleteAlarm();
      await this.ctx.storage.delete('alarmReason');
    }
  }

  async removeSubscriber(userStubName: string): Promise<void> {
    const subs = new Set((await this.ctx.storage.get<string[]>('subs')) ?? []);
    subs.delete(userStubName);
    await this.ctx.storage.put('subs', [...subs]);
    if (subs.size === 0) {
      await this.scheduleGc();
    }
  }

  async subscriberCount(): Promise<number> {
    return ((await this.ctx.storage.get<string[]>('subs')) ?? []).length;
  }

  // ─── Publish ─────────────────────────────────────────────────────────────

  async publish(envelope: Envelope): Promise<void> {
    const coalesced = COALESCED.has(envelope.event ?? '');
    if (coalesced) {
      // Buffer + schedule coalesce window alarm.
      const buf = (await this.ctx.storage.get<Envelope[]>('buf')) ?? [];
      buf.push(envelope);
      await this.ctx.storage.put('buf', buf);
      const al = await this.ctx.storage.getAlarm();
      if (al === null) {
        const ms = await this.coalesceWindowMs();
        await this.ctx.storage.setAlarm(Date.now() + ms);
        await this.ctx.storage.put('alarmReason', 'flush');
      }
      return;
    }
    // Immediate fan-out.
    await this.fanOut([envelope]);
  }

  // ─── Alarm handler ───────────────────────────────────────────────────────

  override async alarm(): Promise<void> {
    const reason = await this.ctx.storage.get<'flush' | 'gc'>('alarmReason');
    await this.ctx.storage.delete('alarmReason');
    if (reason === 'flush') {
      const buf = (await this.ctx.storage.get<Envelope[]>('buf')) ?? [];
      await this.ctx.storage.delete('buf');
      // Deduplicate: one envelope per event id — latest payload wins.
      const byEvent = new Map<string, Envelope>();
      for (const e of buf) byEvent.set(e.event ?? '', e);
      await this.fanOut([...byEvent.values()]);
    } else if (reason === 'gc') {
      const n = ((await this.ctx.storage.get<string[]>('subs')) ?? []).length;
      if (n === 0) await this.ctx.storage.deleteAll();
    }
  }

  // ─── Test helpers ─────────────────────────────────────────────────────────

  /** Exposed for unit test only — no cost in prod (runs once per test invocation). */
  async coalesceWindowMsForTest(): Promise<number> {
    return this.coalesceWindowMs();
  }

  // ─── Private helpers ─────────────────────────────────────────────────────

  private async coalesceWindowMs(): Promise<number> {
    const n = ((await this.ctx.storage.get<string[]>('subs')) ?? []).length;
    if (n > 200) return 2000;
    if (n > 50) return 1000;
    return 500;
  }

  private async fanOut(envs: Envelope[]): Promise<void> {
    const subs = (await this.ctx.storage.get<string[]>('subs')) ?? [];
    if (subs.length === 0) return;
    for (let i = 0; i < subs.length; i += FAN_OUT_BATCH) {
      const slice = subs.slice(i, i + FAN_OUT_BATCH);
      await Promise.allSettled(
        slice.map((name) => {
          const stub = this.env.USER_SESSION_DO.get(this.env.USER_SESSION_DO.idFromName(name));
          return Promise.all(envs.map((e) => stub.deliver(e as never)));
        }),
      );
    }
  }

  private async scheduleGc(): Promise<void> {
    await this.ctx.storage.setAlarm(Date.now() + GC_IDLE_MS);
    await this.ctx.storage.put('alarmReason', 'gc');
  }
}