/**
 * UserSessionDO — per-user WebSocket session terminus.
 *
 * One DO instance per stub name (e.g. "user:uuid", "anon:sid").
 * Accepts all WS connections for a user across tabs/devices.
 * Uses Cloudflare WebSocket Hibernation API via BaseWebSocketDO.
 *
 * Calls loadPrefs / markRead / canSubscribe in-process (neon-http via getDb)
 * — no RPC service bindings required; DO worker already cross-imports from
 * @multideal/web like the cron dispatcher does.
 */
import { BaseWebSocketDO } from './_base-ws.js';
import type { MultidealEnv } from '../lib/env.js';
import { getDb } from '../lib/db.js';
import { loadPrefs, allowsEvent, type NotifPrefs } from '@/server/db/queries/notif-prefs';
import { markRead } from '@/server/db/queries/live-notifications';
import { canSubscribe } from '@/server/notifications/topics';
import { captureCaught } from '@/server/observability/capture.server.js';

// ─── Envelope types ───────────────────────────────────────────────────────────
type Envelope =
  | {
      ch: 'notif';
      kind: 'inbox';
      tier: 'critical' | 'optional' | 'marketing';
      notif: { id: string; event: string; titleHe: string; titleEn: string };
      ts: number;
    }
  | {
      ch: 'notif';
      kind: 'read_sync';
      tier: 'critical';
      ids: string[];
      ts: number;
    }
  | { ch: 'notif'; kind: 'prefs_refresh'; tier: 'critical'; ts: number }
  | {
      ch: string;
      kind: 'live';
      tier: 'critical' | 'optional' | 'marketing';
      event: string;
      payload: unknown;
      ts: number;
      msgId?: string;
    }
  | {
      ch: 'system';
      kind: 'hello';
      tier: 'critical';
      userId: string;
      unreadCount: number;
      serverTime: number;
    }
  | {
      ch: 'system';
      kind: 'error';
      tier: 'critical';
      code: string;
      ch_requested?: string;
    };

// ─── Client ops ──────────────────────────────────────────────────────────────
type ClientOp =
  | { op: 'hello'; lastInboxTs?: number }
  | { op: 'sub'; ch: string }
  | { op: 'unsub'; ch: string }
  | { op: 'read'; ids: string[] }
  | { op: 'typing'; ch: string; isTyping: boolean };

const PREFS_TTL_MS = 5 * 60_000;
const FRAME_RATE_LIMIT_PER_MIN = 120;

export class UserSessionDO extends BaseWebSocketDO<MultidealEnv> {
  // ─── RPC surface (called by TopicDO.fanOut and main worker) ──────────────

  /**
   * Deliver an envelope to all live sockets for this user.
   * Respects tier/pref filtering for optional & marketing tiers.
   * Bumps unreadCount for inbox-kind envelopes.
   */
  async deliver(envelope: Envelope): Promise<void> {
    const sockets = this.liveSockets();
    if (sockets.length === 0) return;

    if (envelope.ch === 'notif' && envelope.kind === 'inbox') {
      const prefs = await this.getPrefs();
      const allowed = allowsEvent(
        prefs,
        envelope.notif.event,
        envelope.tier as 'critical' | 'optional' | 'marketing',
      );
      if (!allowed) return;
      const unread = ((await this.ctx.storage.get<number>('unreadCount')) ?? 0) + 1;
      await this.ctx.storage.put('unreadCount', unread);
    }

    console.warn(
      JSON.stringify({
        event: 'notif.deliver',
        socketCount: sockets.length,
        tier: 'tier' in envelope ? envelope.tier : 'critical',
        channel: envelope.ch,
      }),
    );
    this.broadcast(envelope);
  }

  async hasOpenSockets(): Promise<boolean> {
    return this.liveSockets().length > 0;
  }

  async refreshPrefs(): Promise<void> {
    await this.ctx.storage.delete('prefs');
    await this.ctx.storage.delete('prefsAt');
    this.broadcast({
      ch: 'notif',
      kind: 'prefs_refresh',
      tier: 'critical',
      ts: Date.now(),
    });
  }

  async getStats(): Promise<{
    socketCount: number;
    subscriptions: number;
    unreadCount: number;
  }> {
    const subs = ((await this.ctx.storage.get<string[]>('subscriptions')) ?? []).length;
    const unread = (await this.ctx.storage.get<number>('unreadCount')) ?? 0;
    return {
      socketCount: this.liveSockets().length,
      subscriptions: subs,
      unreadCount: unread,
    };
  }

  /** Test-only helper (preserved in prod but no callers outside tests). */
  async applyTestPrefs(prefs: NotifPrefs): Promise<void> {
    await this.ctx.storage.put('prefs', prefs);
    await this.ctx.storage.put('prefsAt', Date.now());
  }

  // ─── WS upgrade ──────────────────────────────────────────────────────────

  override async fetch(req: Request): Promise<Response> {
    if (req.headers.get('upgrade')?.toLowerCase() !== 'websocket') {
      return new Response('expected WebSocket', { status: 426 });
    }
    const url = new URL(req.url);
    const userId = url.searchParams.get('userId') ?? 'anon';
    const deviceId = url.searchParams.get('device') ?? crypto.randomUUID();
    const role = (url.searchParams.get('role') ?? 'user') as 'user' | 'vendor' | 'admin';
    const [client, server] = Object.values(new WebSocketPair()) as [WebSocket, WebSocket];
    this.acceptHibernated(server, [userId, deviceId, role]);
    const unread = (await this.ctx.storage.get<number>('unreadCount')) ?? 0;
    server.send(
      JSON.stringify({
        ch: 'system',
        kind: 'hello',
        tier: 'critical',
        userId,
        unreadCount: unread,
        serverTime: Date.now(),
      } satisfies Envelope),
    );
    return new Response(null, { status: 101, webSocket: client });
  }

  // ─── Inbound frames ───────────────────────────────────────────────────────

  protected override async onMessage(ws: WebSocket, raw: string | ArrayBuffer): Promise<void> {
    if (!(await this.allowFrame())) {
      try {
        ws.send(
          JSON.stringify({
            ch: 'system',
            kind: 'error',
            tier: 'critical',
            code: 'rate_limit',
          }),
        );
      } catch (err) {
        captureCaught(err, {
          scope: 'server.do-host.user-session.send.rate-limit',
          severity: 'info',
        });
      }
      return;
    }
    const text = typeof raw === 'string' ? raw : new TextDecoder().decode(raw);
    if (text.length > 4 * 1024) {
      try {
        ws.send(
          JSON.stringify({
            ch: 'system',
            kind: 'error',
            tier: 'critical',
            code: 'frame_too_large',
          }),
        );
      } catch (err) {
        captureCaught(err, {
          scope: 'server.do-host.user-session.send.frame-size',
          severity: 'info',
        });
      }
      return;
    }
    let op: ClientOp;
    try {
      op = JSON.parse(text) as ClientOp;
    } catch (err) {
      captureCaught(err, {
        scope: 'server.do-host.user-session.parse-frame',
        severity: 'info',
      });
      return;
    }

    const tags = this.ctx.getTags(ws);
    const [userId, , role] = tags as [string, string, 'user' | 'vendor' | 'admin'];

    switch (op.op) {
      case 'hello': {
        // Reconnect ack — replay delegated to /api/notifications REST; keep DO lean.
        void op;
        return;
      }
      case 'sub': {
        const db = getDb(this.env);
        const ok = await canSubscribe(db, userId, op.ch, role);
        if (!ok) {
          try {
            ws.send(
              JSON.stringify({
                ch: 'system',
                kind: 'error',
                tier: 'critical',
                code: 'channel_forbidden',
                ch_requested: op.ch,
              }),
            );
          } catch (err) {
            captureCaught(err, {
              scope: 'server.do-host.user-session.send.forbidden',
              severity: 'info',
            });
          }
          return;
        }
        const subs = new Set((await this.ctx.storage.get<string[]>('subscriptions')) ?? []);
        if (!subs.has(op.ch)) {
          subs.add(op.ch);
          await this.ctx.storage.put('subscriptions', [...subs]);
          const stub = this.env.TOPIC_DO.get(this.env.TOPIC_DO.idFromName(`topic:${op.ch}`));
          await stub.addSubscriber(`user:${userId}`);
        }
        return;
      }
      case 'unsub': {
        const subs = new Set((await this.ctx.storage.get<string[]>('subscriptions')) ?? []);
        if (subs.delete(op.ch)) {
          await this.ctx.storage.put('subscriptions', [...subs]);
          const stub = this.env.TOPIC_DO.get(this.env.TOPIC_DO.idFromName(`topic:${op.ch}`));
          await stub.removeSubscriber(`user:${userId}`);
        }
        return;
      }
      case 'read': {
        const db = getDb(this.env);
        const ids = [...new Set(op.ids)].slice(0, 100);
        const markedCount = await markRead(db, userId, ids);
        const next = Math.max(
          0,
          ((await this.ctx.storage.get<number>('unreadCount')) ?? 0) - markedCount,
        );
        await this.ctx.storage.put('unreadCount', next);
        this.broadcast({
          ch: 'notif',
          kind: 'read_sync',
          tier: 'critical',
          ids,
          ts: Date.now(),
        });
        return;
      }
      case 'typing': {
        const db = getDb(this.env);
        const ok = await canSubscribe(db, userId, op.ch, role);
        if (!ok) {
          try {
            ws.send(
              JSON.stringify({
                ch: 'system',
                kind: 'error',
                tier: 'critical',
                code: 'channel_forbidden',
                ch_requested: op.ch,
              }),
            );
          } catch (err) {
            captureCaught(err, {
              scope: 'server.do-host.user-session.send.typing-forbidden',
              severity: 'info',
            });
          }
          return;
        }
        const stub = this.env.TOPIC_DO.get(this.env.TOPIC_DO.idFromName(`topic:${op.ch}`));
        await stub.publish({
          ch: op.ch,
          kind: 'live',
          tier: 'optional',
          event: 'chat.typing',
          payload: { userId, isTyping: op.isTyping },
          ts: Date.now(),
        });
        return;
      }
    }
  }

  protected override async onClose(_ws: WebSocket): Promise<void> {
    // Subscriptions stay; if this was the last socket, next deliver no-ops to dead set.
    // TopicDO removal happens on explicit unsub or topic GC.
  }

  // ─── Helpers ─────────────────────────────────────────────────────────────

  private async getPrefs(): Promise<NotifPrefs> {
    const ts = (await this.ctx.storage.get<number>('prefsAt')) ?? 0;
    if (Date.now() - ts < PREFS_TTL_MS) {
      return (await this.ctx.storage.get<NotifPrefs>('prefs')) ?? {};
    }
    const sockets = this.liveSockets();
    const userId = sockets.length > 0 ? (this.ctx.getTags(sockets[0]!)[0] ?? 'anon') : 'anon';
    if (userId === 'anon' || userId.startsWith('anon:')) return {};
    const db = getDb(this.env);
    const prefs = await loadPrefs(db, userId);
    await this.ctx.storage.put('prefs', prefs);
    await this.ctx.storage.put('prefsAt', Date.now());
    return prefs;
  }

  private async allowFrame(): Promise<boolean> {
    const key = `rl:${Math.floor(Date.now() / 60_000)}`;
    const n = ((await this.ctx.storage.get<number>(key)) ?? 0) + 1;
    await this.ctx.storage.put(key, n);
    return n <= FRAME_RATE_LIMIT_PER_MIN;
  }
}