/**
 * @factory/shared/realtime — the WS frame contract (TASK-0091 R3).
 *
 * The single discriminated union for `apps/realtime` (server) and
 * `packages/collaboration` (client mirror) — both sides import the Zod
 * schemas from here; the client's structural types in
 * `packages/collaboration/src/types.ts` are a lockstep mirror and must
 * never drift (the codec/provider tests pin the shape).
 *
 * Framing (recorded in the task file, notes 12/13):
 *  - **Text frames**: one JSON object per frame, `type` discriminant.
 *  - **Binary frames**: 1-byte kind prefix + raw Yjs bytes.
 *      `0x01` — doc update (`Y.applyUpdate`-compatible)
 *      `0x02` — awareness update (y-protocols `applyAwarenessUpdate`)
 *
 * Base64 Yjs payloads live INSIDE the JSON control frames
 * (`sync.request.stateVector`, `sync.state.chunk`); the `update` /
 * `awareness` frames are binary (prefix + raw bytes).
 *
 * Close codes: `4000` protocol/heartbeat, `4001` unauthorized, `4002`
 * forbidden, `4008` rate/policy, `4009` payload, `4010` server shutdown
 * (the scaffold's 44xx set is superseded — task header note 3).
 */

import { z } from "zod";

// --- close codes (TASK-0091 R6) ----------------------------------------------

export const WS_CLOSE = {
  /** Protocol error / heartbeat timeout (no pong in 90 s). */
  ProtocolOrHeartbeat: 4000,
  /** Unauthorized: invalid/expired JWT (no room join happened). */
  Unauthorized: 4001,
  /** Forbidden: wrong workspace / not the socket's own control room / no VIEW. */
  Forbidden: 4002,
  /** Rate limit / policy violation (connect rate, msg rate, 10 strikes). */
  RateOrPolicy: 4008,
  /** Frame larger than the 256 KB cap. */
  PayloadTooLarge: 4009,
  /** Graceful server shutdown (SIGTERM drain — retry after 5 s). */
  ServerShutdown: 4010,
} as const;

export type WsCloseCode = (typeof WS_CLOSE)[keyof typeof WS_CLOSE];

// --- binary framing ------------------------------------------------------------

/** Binary frame kind prefix (the first byte of a binary WS frame). */
export const BINARY_KIND = {
  Update: 0x01,
  Awareness: 0x02,
} as const;

export type BinaryKind = (typeof BINARY_KIND)[keyof typeof BINARY_KIND];

// --- client → server (JSON text frames) ----------------------------------------

/** Join-sync request. Empty `stateVector` = full doc; set = state-vector diff. */
export const SyncRequestFrameSchema = z.object({
  type: z.literal("sync.request"),
  /** Base64 state vector; omitted = the local doc is empty (full sync). */
  stateVector: z.string().optional(),
  /** Last acked server seq (the Redis update-log resume fast path, R5). */
  lastSeq: z.number().int().nonnegative().optional(),
});

/**
 * Application-level heartbeat. Carries the client's last-ack seq so a
 * reconnect can use the update-log resume even between pings (R5).
 */
export const PingFrameSchema = z.object({
  type: z.literal("ping"),
  ts: z.number(),
  lastSeq: z.number().int().nonnegative().optional(),
});

/**
 * Non-editor mutation count pending in the local outbox (sent after `hello`
 * on (re)connect so the server can estimate the drain load).
 */
export const OutboxPendingFrameSchema = z.object({
  type: z.literal("outbox.pending"),
  count: z.number().int().nonnegative(),
});

/**
 * Control-channel join (TASK-0092 (b) / 6b): open a page's control room on
 * THIS socket (the web `RealtimeClient` is one-socket-many-channels — it
 * connects token-only, no `?room=`, and opens channels in-band). `page:<id>`
 * channels receive the server's control frames (the `block:changed`
 * refetch-hint; comments/notifications in TASK-0094); `workspace:<id>` is
 * accepted (the web client may (re)subscribe it) but no frames ride it yet.
 * Idempotent — re-subscribing an open channel is a no-op (reconnect).
 */
export const SubscribeFrameSchema = z.object({
  type: z.literal("subscribe"),
  channel: z
    .string()
    .refine((c) => /^page:[^:\s]{1,}$/.test(c) || /^workspace:[^:\s]{1,}$/.test(c), {
      message: "channel must be page:<id> or workspace:<id>",
    }),
});

/** Control-channel leave (the inverse of {@link SubscribeFrameSchema}). */
export const UnsubscribeFrameSchema = z.object({
  type: z.literal("unsubscribe"),
  channel: z
    .string()
    .refine((c) => /^page:[^:\s]{1,}$/.test(c) || /^workspace:[^:\s]{1,}$/.test(c), {
      message: "channel must be page:<id> or workspace:<id>",
    }),
});

/**
 * A peer's cursor anchor (doc §5.1): a text offset inside one block.
 * The `selection` sub-object carries a (possibly multi-block) range so the
 * remote tab can render a translucent highlight (TASK-0093 R4).
 */
export const PresenceCursorSchema = z.object({
  blockId: z.string().min(1),
  offset: z.number().int().nonnegative(),
  selection: z
    .object({
      fromBlockId: z.string().min(1),
      fromOffset: z.number().int().nonnegative(),
      toBlockId: z.string().min(1),
      toOffset: z.number().int().nonnegative(),
    })
    .optional(),
});

/**
 * Presence UPDATE (client → server, TASK-0093): the sender's cursor /
 * viewing state. The peer identity is NEVER carried by the client — the
 * server stamps `user` (id/name/color) from the socket's authenticated
 * handshake + the shared `presenceColor` hash before broadcasting (the
 * server-stamped-names decision: a client cannot impersonate a name via
 * presence; the frame is pure state). `cursor: null` clears the cursor
 * (tab hidden / blur → `viewing: false`); `cursor` absent = no change.
 * Rides both socket kinds: the doc-room socket (the TASK-0136 provider)
 * and the roomless control socket over its `page:<id>` channel (the (b)
 * web client).
 */
export const PresenceFrameSchema = z.object({
  type: z.literal("presence"),
  cursor: PresenceCursorSchema.nullable().optional(),
  viewing: z.boolean().optional(),
});

/** Presence LEAVE (client → server): the explicit "I'm done" (page
 *  unmount / `dispose`) — the server drops the peer immediately instead
 *  of waiting for the 90 s prune. Best-effort: a dropped leave is healed
 *  by the prune + the next join's re-broadcast (doc §5.2). */
export const PresenceLeaveFrameSchema = z.object({
  type: z.literal("presence-leave"),
});

/** JSON client frames. Binary client frames: `update` / `awareness` (prefixed). */
export const ClientFrameSchema = z.discriminatedUnion("type", [
  SyncRequestFrameSchema,
  PingFrameSchema,
  OutboxPendingFrameSchema,
  SubscribeFrameSchema,
  UnsubscribeFrameSchema,
  PresenceFrameSchema,
  PresenceLeaveFrameSchema,
]);

// --- server → client (JSON text frames) -----------------------------------------

/** A roster peer (the `hello` payload — users only, no cursors; cursors
 *  arrive via the join re-broadcast + `presence` frames, TASK-0093 R2). */
export const RoomPeerSchema = z.object({
  user: z.object({ id: z.string(), name: z.string(), color: z.string() }),
});

/**
 * Emitted on a successful room join (R1): identity + capability + clock.
 * `canEdit` is the defense-in-depth flag the client also enforces locally
 * (a VIEW-only peer's local updates are dropped, mirroring the server).
 */
export const HelloFrameSchema = z.object({
  type: z.literal("hello"),
  connId: z.string(),
  canEdit: z.boolean(),
  serverTime: z.number(),
  roomPeers: z.array(RoomPeerSchema),
});

/**
 * Paged doc state: one base64 Yjs update per chunk (≤ 64 KB server-side),
 * `done` marks the end of the transfer (R3; paged hydration, TASK-0092).
 */
export const SyncStateFrameSchema = z.object({
  type: z.literal("sync.state"),
  chunk: z.string(),
  done: z.boolean(),
});

/**
 * Server seq ack for a logged update. When the server sends no acks the
 * client falls back to pure state-vector resume (always correct — the log
 * is a speed optimization only, R5).
 */
export const AckFrameSchema = z.object({
  type: z.literal("ack"),
  seq: z.number().int().nonnegative(),
});

/** A persisted doc state reached Postgres (the "Saved" indicator). */
export const DocPersistedFrameSchema = z.object({
  type: z.literal("doc.persisted"),
  pageId: z.string(),
  version: z.number().int(),
});

/**
 * Page-block change REFETCH HINT over the `page:{pageId}` control room
 * (TASK-0092 R6 / (b) — the user-chosen additive bridge). The producer
 * (the realtime server) pushes this on a `page.updated` domain event
 * (REST write or a flush) instead of replaying per-block patches: a web
 * `Block` carries `createdAt`/`updatedAt` (rendered in the context menu +
 * page footer) which the flush's minimal `BlockOperation`s do NOT, so a
 * lossless `block:upsert` would need a DB row read per block. The hint
 * says "this page's blocks changed → refetch" (the client re-reads the DB,
 * the source of truth at rest) — lossless, correct timestamps, no per-
 * block fan-out. `version` (the new `pages.version`) is advisory: the
 * client may use it for dedup/last-write-wins but MUST NOT gate on it.
 */
export const BlockChangedFrameSchema = z.object({
  type: z.literal("block:changed"),
  pageId: z.string().min(1),
  /** The new `pages.version` (advisory — refetch anyway). */
  version: z.number().int().min(1).optional(),
});

/**
 * Presence BROADCAST (server → client, TASK-0093): the full current state
 * of ONE peer in a page/database room. The server stamps `user`
 * (server-stamped names — id from the socket's handshake, name from the
 * identity resolver, color from the shared `presenceColor` hash); the
 * client never supplies it. Fans out on:
 *   - a peer's `presence` frame (its cursor/viewing update),
 *   - a room's JOIN re-broadcast (every peer's state, cursors included —
 *     the deterministic catch-up for joiners, doc §5.1),
 *   - a prune (the pruned peer is removed via `presence-leave`),
 *   - > 50 peers (the cap): `cursor`/`viewing` are stripped, `user` stays
 *     (avatars-only mode, doc §5.1).
 * Over doc-room sockets the frame carries the room's id in `pageId` (a
 * database room carries `databaseId` in the same field); over roomless
 * control sockets it rides the subscribed `page:<id>` channel (the (b)
 * web client — one socket, many channels).
 */
export const PresenceBroadcastSchema = z.object({
  type: z.literal("presence"),
  /** The room the state belongs to (`page:<id>` or `database:<id>`). */
  pageId: z.string().min(1),
  /** The peer's server-stamped identity. */
  user: z.object({ id: z.string().min(1), name: z.string(), color: z.string() }),
  /** The peer's cursor (null = cleared / hidden / avatars-only cap). */
  cursor: PresenceCursorSchema.nullable().optional(),
  /** Whether the peer is actively viewing (false = background tab — the
   *  cursor is hidden, the avatar stays). */
  viewing: z.boolean().optional(),
  /** Server epoch ms (the client's prune basis + join-latency metric). */
  lastSeen: z.number().int().nonnegative(),
});

/**
 * Presence LEAVE broadcast (server → client, TASK-0093): the peer left
 * the room (explicit `presence-leave`, the socket closed, or the 90 s
 * prune). The client removes the peer from its roster (the bar updates,
 * the "X left" toast fires). Best-effort — a missed leave is healed by
 * the client's own prune window + the next join's re-broadcast (doc §5.2).
 */
export const PresenceLeaveBroadcastSchema = z.object({
  type: z.literal("presence-leave"),
  /** The room the peer left (`page:<id>` or `database:<id>`). */
  pageId: z.string().min(1),
  userId: z.string().min(1),
});

/** Comment fan-out over the page room (TASK-0094 R1). */
export const CommentEventFrameSchema = z.object({
  type: z.literal("comment.event"),
  kind: z.enum(["created", "replied", "resolved", "reopened", "reacted", "deleted"]),
  commentId: z.string(),
  pageId: z.string(),
  actorId: z.string(),
  /** First ~200 chars + render hint (the full comment is fetched on demand). */
  preview: z.string(),
});

/** Notification push over the `user:{userId}` control room (TASK-0094 R7). */
export const NotificationPushFrameSchema = z.object({
  type: z.literal("notification.push"),
  notificationId: z.string(),
  /** The notification kind (comment.reply, mention, …) — `type` is the
   *  frame discriminator, so the payload field is `kind`. */
  kind: z.string(),
  title: z.string(),
  deepLink: z.object({ type: z.string(), id: z.string() }),
  actorName: z.string().optional(),
});

/** Long-hydration notice (the client shows "Loading page…", TASK-0092 R2). */
export const HydratingFrameSchema = z.object({
  type: z.literal("hydrating"),
  pageId: z.string(),
});

/** Protocol error (a close is NOT implied — the socket stays open). */
export const ErrorFrameSchema = z.object({
  type: z.literal("error"),
  code: z.string(),
  message: z.string(),
});

/** Heartbeat echo. */
export const PingPongFrameSchema = z.object({
  type: z.literal("ping-pong"),
  ts: z.number(),
});

export const ServerFrameSchema = z.discriminatedUnion("type", [
  HelloFrameSchema,
  SyncStateFrameSchema,
  AckFrameSchema,
  DocPersistedFrameSchema,
  BlockChangedFrameSchema,
  PresenceBroadcastSchema,
  PresenceLeaveBroadcastSchema,
  CommentEventFrameSchema,
  NotificationPushFrameSchema,
  HydratingFrameSchema,
  ErrorFrameSchema,
  PingPongFrameSchema,
]);

// --- inferred types (client mirror + server handlers) ---------------------------

export type SyncRequestFrame = z.infer<typeof SyncRequestFrameSchema>;
export type PingFrame = z.infer<typeof PingFrameSchema>;
export type OutboxPendingFrame = z.infer<typeof OutboxPendingFrameSchema>;
export type SubscribeFrame = z.infer<typeof SubscribeFrameSchema>;
export type UnsubscribeFrame = z.infer<typeof UnsubscribeFrameSchema>;
export type PresenceCursor = z.infer<typeof PresenceCursorSchema>;
export type PresenceFrame = z.infer<typeof PresenceFrameSchema>;
export type PresenceLeaveFrame = z.infer<typeof PresenceLeaveFrameSchema>;
export type ClientFrame = z.infer<typeof ClientFrameSchema>;

export type RoomPeer = z.infer<typeof RoomPeerSchema>;
export type HelloFrame = z.infer<typeof HelloFrameSchema>;
export type SyncStateFrame = z.infer<typeof SyncStateFrameSchema>;
export type AckFrame = z.infer<typeof AckFrameSchema>;
export type DocPersistedFrame = z.infer<typeof DocPersistedFrameSchema>;
export type BlockChangedFrame = z.infer<typeof BlockChangedFrameSchema>;
export type PresenceBroadcast = z.infer<typeof PresenceBroadcastSchema>;
export type PresenceLeaveBroadcast = z.infer<typeof PresenceLeaveBroadcastSchema>;
export type CommentEventFrame = z.infer<typeof CommentEventFrameSchema>;
export type NotificationPushFrame = z.infer<typeof NotificationPushFrameSchema>;
export type HydratingFrame = z.infer<typeof HydratingFrameSchema>;
export type ErrorFrame = z.infer<typeof ErrorFrameSchema>;
export type PingPongFrame = z.infer<typeof PingPongFrameSchema>;
export type ServerFrame = z.infer<typeof ServerFrameSchema>;

/** One decoded wire frame (text = JSON object, binary = prefixed Yjs bytes). */
export type WireFrame =
  | { readonly kind: "text"; readonly frame: ClientFrame | ServerFrame }
  | {
      readonly kind: "binary";
      readonly binary: "update" | "awareness";
      readonly bytes: Uint8Array;
    };

// --- close-code hints (the client toast text) ------------------------------------

/** Close code → user-facing reconnect hint (the client toast text). */
export const CLOSE_HINTS: Readonly<Record<number, string>> = {
  [WS_CLOSE.ProtocolOrHeartbeat]: "Connection lost — reconnecting…",
  [WS_CLOSE.Unauthorized]: "Session expired — signing you back in…",
  [WS_CLOSE.Forbidden]: "You no longer have access to this page.",
  [WS_CLOSE.RateOrPolicy]: "Too many changes — giving the server a moment…",
  [WS_CLOSE.PayloadTooLarge]: "A change was too large to send — retrying…",
  [WS_CLOSE.ServerShutdown]: "The server is restarting — reconnecting in a moment…",
};

// --- server-side limit constants (TASK-0091, doc §2.1/§6.2) -----------------------

export const REALTIME_LIMITS = {
  /** New connections per IP per minute (connect-time limiter). */
  connectPerIpPerMin: 30,
  /** Live sockets per user (global, the server counts live sockets). */
  maxConnsPerUser: 100,
  /** Max frame size in bytes (a bigger frame → close 4009). */
  maxFrameBytes: 256 * 1024,
  /** Sustained messages per socket per second (token bucket). */
  maxMsgsPerSocketPerSec: 100,
  /** Server→client sync.state chunk cap (R3 paged transfer). */
  maxSyncChunkBytes: 64 * 1024,
  /** Update log: last N entries (Redis list cap, R5). */
  updateLogMaxEntries: 2000,
  /** Update log: total byte cap (R5, whichever bound hits first). */
  updateLogMaxBytes: 10 * 1024 * 1024,
  /** Server ping cadence (ms, R4). */
  pingIntervalMs: 30_000,
  /** Silence before a socket is pruned with 4000 (ms, R4). */
  pruneSilenceMs: 90_000,
  /** Awareness silence before a peer's state is pruned (ms, doc §5.1). */
  awarenessPruneMs: 90_000,
  /** Presence fan-out cap: > this many peers → avatars only, no cursors. */
  awarenessPeerCap: 50,
  /** Live control sockets per user (multi-tab abuse guard, note 16). */
  maxControlSocketsPerUser: 10,
  /** Bad frames in 10 s before the socket is dropped (R3 10-strike). */
  strikeLimit: 10,
  /** The strike window (ms, R3). */
  strikeWindowMs: 10_000,
  /** Empty-room eviction (the "warm room" lifetime, R2). */
  roomGcMs: 5 * 60_000,
  /** Max buffered updates while a room is hydrating (doc §4.2). */
  hydrateBufferLimit: 5_000,
  /** Hydration deadline (ms; buffer flush + error on overflow, doc §4.2). */
  hydrateMaxMs: 5_000,
  /** SIGTERM drain budget (ms, R6). */
  drainMaxMs: 10_000,
  /** retryAfterMs advertised on the 4010 close frame (R6). */
  shutdownRetryAfterMs: 5_000,
} as const;