/**
 * Columns persistence (P1 — design §1.2/§1.4).
 *
 * `stream_columns` (021) is the persisted column list; this repository owns
 * every mutation plus the resolved-source rows in `column_sources` (026) and
 * the per-source item rows in `feed_items`/`feed_item_columns` written by the
 * poll loop. Reads run under `withTenant` so FORCE RLS binds honestly.
 *
 * Everything the client can send is parsed with the P0a core validators
 * (`parseStreamColumnSpec`, `parseRefreshPolicyClamped`, `clampTitle`,
 * `clampPosition`) — the DB CHECKs are backstops, never the validation layer.
 * `StreamColumnsValidationError` is translated to 422 at the controller.
 */
import { BadRequestException, Inject, Injectable, NotFoundException, Optional, UnprocessableEntityException } from "@nestjs/common";
import { createHash } from "node:crypto";
import type { PoolClient } from "pg";
import {
  clampPosition, clampTitle, parseRefreshPolicyClamped, parseStreamColumnSpec,
  StreamColumnsValidationError, type RefreshPolicy, type StreamColumnSpec, type StreamColumnType,
} from "@tactical-media-broadcast-hub/core";
import type { SessionPayload } from "../auth/session.js";
import { withTenant } from "../db/client.js";
import { appendTenantAudit } from "../audit/audit.service.js";
import type { NormalizedFeedItem } from "./feed-normalize.js";
import type { ConnectionSummary } from "../connections/connections.service.js";

export type ColumnSourceKind = "home" | "mentions" | "search" | "author";
export type ColumnSourceState =
  | "pending" | "ready" | "stale" | "error" | "unsupported" | "rate_limited" | "budget_paused" | "disabled";

export interface ColumnRecord {
  id: string;
  type: StreamColumnType;
  title: string;
  position: number;
  spec: StreamColumnSpec;
  refreshPolicy: RefreshPolicy;
  createdAt: string;
  updatedAt: string;
}

/** `ColumnView.source` (design §1.4): the scheduler-facing row, squashed. */
export interface ColumnSourceView {
  id: string;
  /** Owning column — present so a bulk source query can be grouped per column. */
  columnId: string;
  provider: string;
  connectionId: string | null;
  sourceKind: ColumnSourceKind;
  query: string | null;
  actorRef: string | null;
  state: ColumnSourceState;
  lastError: string | null;
  lastPolledAt: string | null;
  nextPollAt: string | null;
  consecutiveFailures: number;
}

export interface CreateColumn {
  type: StreamColumnType;
  title: string;
  position?: number;
  spec: unknown;
  refreshPolicy?: unknown;
}

export interface UpdateColumn {
  title?: string;
  position?: number;
  spec?: unknown;
  refreshPolicy?: unknown;
}

export interface StoredFeedItem {
  id: string;
  dedupKey: string;
  createdAtRemote: string;
  item: NormalizedFeedItem;
}

/** The identity one poll resolves to — see `sourceIdentities` below. */
interface SourceIdentity {
  provider: string;
  connectionId: string | null;
  sourceKind: ColumnSourceKind;
  query: string | null;
  actorRef: string | null;
}

const identityKey = (identity: SourceIdentity): string =>
  JSON.stringify([identity.provider, identity.connectionId, identity.sourceKind, identity.query, identity.actorRef]);

export function mapValidationError<T>(parse: () => T): T {
  try {
    return parse();
  } catch (error) {
    if (error instanceof StreamColumnsValidationError) throw new UnprocessableEntityException(error.message);
    throw error;
  }
}

/**
 * sha256(workspace | provider | canonical remote id) — design §3.3. The
 * canonicalization keeps a provider that restates an item with decorations
 * (e.g. Mastodon's `scheduled:123` vs `123`) from double-storing the same
 * remote object.
 */
export function dedupKeyFor(workspaceId: string, provider: string, remoteId: string): string {
  const canonical = remoteId.replace(/^scheduled:/, "");
  return createHash("sha256").update(`${workspaceId}|${provider}|${canonical}`).digest("hex");
}

/**
 * The polled identities a column spec resolves to (§3.3/§3.4). platform_feed
 * is the degenerate case: one connection, one kind per source row. keyword and
 * saved_search expand to one keyless search source per platform. account_list
 * expands to one author source per account entry.
 */
export function sourceIdentities(type: StreamColumnType, spec: StreamColumnSpec): SourceIdentity[] {
  switch (type) {
    case "platform_feed": {
      const feed = spec as import("@tactical-media-broadcast-hub/core").PlatformFeedSpec;
      const kinds = feed.kinds.length > 0 ? feed.kinds : (["home"] as const);
      return kinds.map((kind): SourceIdentity => ({
        provider: "", connectionId: feed.connectionId, sourceKind: kind, query: null, actorRef: null,
      }));
    }
    case "keyword":
    case "saved_search": {
      const keyword = spec as import("@tactical-media-broadcast-hub/core").KeywordSpec;
      return keyword.platforms.map((provider): SourceIdentity => ({
        provider, connectionId: null, sourceKind: "search", query: keyword.query, actorRef: null,
      }));
    }
    case "account_list": {
      const list = spec as import("@tactical-media-broadcast-hub/core").AccountListSpec;
      return list.accounts.map((account): SourceIdentity => ({
        provider: account.provider, connectionId: null, sourceKind: "author", query: null, actorRef: account.handle,
      }));
    }
  }
}

/** Providers referenced by a spec (empty for platform_feed: the connection supplies it). */
export function columnProviders(type: StreamColumnType, spec: StreamColumnSpec): string[] {
  switch (type) {
    case "platform_feed": return [];
    case "keyword":
    case "saved_search": return (spec as import("@tactical-media-broadcast-hub/core").KeywordSpec).platforms;
    case "account_list": return (spec as import("@tactical-media-broadcast-hub/core").AccountListSpec).accounts.map((account) => account.provider);
  }
}

/** The title today's implicit column renders: "Display Name · handle" (web streams-data liveColumns). */
export function seedTitle(row: ConnectionSummary): string {
  const displayName = String(row.fields.displayName ?? "").trim() || row.channelId;
  const handle = String(row.fields.handle ?? "").trim() || row.channelId;
  return clampTitle(`${displayName} · ${handle}`);
}

interface ColumnRow {
  id: string; type: StreamColumnType; title: string; position: number;
  spec: unknown; refresh_policy: unknown; created_at: Date; updated_at: Date;
}

const COLUMN_SELECT = "id, type, title, position, spec, refresh_policy, created_at, updated_at";

function toRecord(row: ColumnRow): ColumnRecord {
  return {
    id: row.id,
    type: row.type,
    title: row.title,
    position: row.position,
    spec: mapValidationError(() => parseStreamColumnSpec(row.type, row.spec)),
    refreshPolicy: parseRefreshPolicyClamped(row.refresh_policy ?? {}),
    createdAt: row.created_at.toISOString(),
    updatedAt: row.updated_at.toISOString(),
  };
}

const SOURCE_SELECT = `SELECT s.id, s.column_id, s.provider, s.connection_id, s.source_kind, s.query, s.actor_ref,
  s.state, s.last_error, s.last_polled_at, s.next_poll_at, s.consecutive_failures
  FROM column_sources s WHERE s.workspace_id = $1`;

function sourceFromRow(row: Record<string, unknown>): ColumnSourceView {
  return {
    id: String(row.id),
    columnId: String(row.column_id),
    provider: String(row.provider),
    connectionId: row.connection_id === null || row.connection_id === undefined ? null : String(row.connection_id),
    sourceKind: row.source_kind as ColumnSourceKind,
    query: row.query === null || row.query === undefined ? null : String(row.query),
    actorRef: row.actor_ref === null || row.actor_ref === undefined ? null : String(row.actor_ref),
    state: row.state as ColumnSourceState,
    lastError: row.last_error === null || row.last_error === undefined ? null : String(row.last_error),
    lastPolledAt: isoOrNull(row.last_polled_at),
    nextPollAt: isoOrNull(row.next_poll_at),
    consecutiveFailures: Number(row.consecutive_failures ?? 0),
  };
}

function isoOrNull(value: unknown): string | null {
  return value === null || value === undefined ? null : (value as Date).toISOString();
}

export interface ColumnStoreDependencies {
  /** Resolve a connection's provider for platform_feed source rows. */
  providerOf(workspaceId: string, connectionId: string): Promise<string>;
  /** Feed-capable connection summaries — the lazy seeding input. */
  connections(workspaceId: string): Promise<ConnectionSummary[]>;
}

export const COLUMN_STORE_DEPENDENCIES = Symbol("COLUMN_STORE_DEPENDENCIES");

@Injectable()
export class ColumnStore {
  constructor(@Optional() @Inject(COLUMN_STORE_DEPENDENCIES) private readonly deps?: ColumnStoreDependencies) {}

  private async providerOf(workspaceId: string, connectionId: string): Promise<string> {
    if (this.deps) return this.deps.providerOf(workspaceId, connectionId);
    return withTenant(workspaceId, async (client): Promise<string> => {
      const result = await client.query<{ provider: string }>(
        "SELECT provider FROM connections WHERE id = $1 AND workspace_id = $2 AND disconnected_at IS NULL", [connectionId, workspaceId]);
      const row = result.rows[0];
      if (row === undefined) throw new NotFoundException(`connection ${connectionId} not found`);
      return row.provider;
    });
  }

  private async connections(workspaceId: string): Promise<ConnectionSummary[]> {
    if (this.deps) return this.deps.connections(workspaceId);
    const { PgConnectionStore } = await import("../connections/connection-store.js");
    return new PgConnectionStore().list(workspaceId);
  }

  async list(workspaceId: string): Promise<ColumnRecord[]> {
    return withTenant(workspaceId, async (client) => {
      const result = await client.query(
        `SELECT ${COLUMN_SELECT} FROM stream_columns WHERE deleted_at IS NULL ORDER BY position, created_at`);
      return result.rows.map((row) => toRecord(row as ColumnRow));
    });
  }

  async get(workspaceId: string, id: string): Promise<ColumnRecord> {
    return withTenant(workspaceId, async (client) => this.read(client, workspaceId, id));
  }

  async sources(workspaceId: string, columnId?: string): Promise<ColumnSourceView[]> {
    return withTenant(workspaceId, async (client) => {
      const result = columnId === undefined
        ? await client.query(`${SOURCE_SELECT} ORDER BY s.created_at`, [workspaceId])
        : await client.query(`${SOURCE_SELECT} AND s.column_id = $2 ORDER BY s.created_at`, [workspaceId, columnId]);
      return result.rows.map((row) => sourceFromRow(row));
    });
  }

  async create(session: SessionPayload, input: CreateColumn): Promise<ColumnRecord> {
    const spec = mapValidationError(() => parseStreamColumnSpec(input.type, input.spec));
    const policy = parseRefreshPolicyClamped(input.refreshPolicy ?? {});
    const title = clampTitle(input.title ?? "");
    if (title === "") throw new UnprocessableEntityException("title must be 1..80 characters after trimming");
    return withTenant(session.workspaceId, async (client) => {
      await this.assertSpecReferenceable(client, session.workspaceId, input.type, spec);
      const position = input.position === undefined
        ? await this.nextPosition(client, session.workspaceId)
        : clampPosition(input.position);
      const result = await client.query(
        `INSERT INTO stream_columns (workspace_id, type, title, position, spec, refresh_policy, created_by)
         VALUES ($1,$2,$3,$4,$5,$6,$7) RETURNING ${COLUMN_SELECT}`,
        [session.workspaceId, input.type, title, position, JSON.stringify(spec), JSON.stringify(policy), session.userId],
      );
      const record = toRecord(result.rows[0] as ColumnRow);
      await this.writeSources(client, session.workspaceId, record.id, input.type, spec);
      await appendTenantAudit(session.workspaceId, {
        actorUserId: session.userId, actorRole: session.role, action: "columns.created",
        targetType: "stream_column", targetId: record.id, diff: { type: input.type, title },
      });
      return record;
    });
  }

  async update(session: SessionPayload, id: string, input: UpdateColumn): Promise<ColumnRecord> {
    if (Object.keys(input).length === 0) throw new BadRequestException("no recognized fields to update");
    return withTenant(session.workspaceId, async (client) => {
      await client.query("SELECT id FROM stream_columns WHERE workspace_id = $1 AND id = $2 AND deleted_at IS NULL FOR UPDATE", [session.workspaceId, id]);
      const current = await this.read(client, session.workspaceId, id);
      const title = input.title === undefined ? current.title : clampTitle(input.title);
      if (title === "") throw new UnprocessableEntityException("title must be 1..80 characters after trimming");
      // Column type is immutable (design §1.4: PATCH carries title/position/
      // spec/refresh_policy); a different type is a new column.
      const spec = input.spec === undefined ? current.spec : mapValidationError(() => parseStreamColumnSpec(current.type, input.spec));
      const policy = input.refreshPolicy === undefined ? current.refreshPolicy : parseRefreshPolicyClamped(input.refreshPolicy);
      const position = input.position === undefined ? current.position : clampPosition(input.position);
      if (input.spec !== undefined) await this.assertSpecReferenceable(client, session.workspaceId, current.type, spec);
      const result = await client.query(
        `UPDATE stream_columns SET title = $3, position = $4, spec = $5, refresh_policy = $6, updated_at = now()
         WHERE workspace_id = $1 AND id = $2 AND deleted_at IS NULL RETURNING ${COLUMN_SELECT}`,
        [session.workspaceId, id, title, position, JSON.stringify(spec), JSON.stringify(policy)],
      );
      if (result.rows[0] === undefined) throw new NotFoundException(`column ${id} not found`);
      if (input.spec !== undefined) await this.rewriteSources(client, session.workspaceId, id, current.type, spec);
      await appendTenantAudit(session.workspaceId, {
        actorUserId: session.userId, actorRole: session.role, action: "columns.updated",
        targetType: "stream_column", targetId: id,
        diff: { titleChanged: title !== current.title, positionChanged: position !== current.position, specChanged: input.spec !== undefined },
      });
      return toRecord(result.rows[0] as ColumnRow);
    });
  }

  private async writeSources(client: PoolClient, workspaceId: string, columnId: string, type: StreamColumnType, spec: StreamColumnSpec): Promise<void> {
    for (const identity of sourceIdentities(type, spec)) {
      const provider = identity.provider !== "" ? identity.provider : await this.providerOf(workspaceId, identity.connectionId ?? "");
      await client.query(
        `INSERT INTO column_sources (workspace_id, column_id, provider, connection_id, source_kind, query, actor_ref)
         VALUES ($1,$2,$3,$4,$5,$6,$7)
         ON CONFLICT (workspace_id, column_id, provider, source_kind, connection_id, query, actor_ref) DO NOTHING`,
        [workspaceId, columnId, provider, identity.connectionId, identity.sourceKind, identity.query, identity.actorRef],
      );
    }
  }

  /** Spec identity changes rewrite the source set; identities no longer referenced by this column are dropped. */
  private async rewriteSources(client: PoolClient, workspaceId: string, columnId: string, type: StreamColumnType, spec: StreamColumnSpec): Promise<void> {
    const wanted = new Set(sourceIdentities(type, spec).map(identityKey));
    const existing = await client.query<{ id: string; provider: string; connection_id: string | null; source_kind: ColumnSourceKind; query: string | null; actor_ref: string | null }>(
      "SELECT id, provider, connection_id, source_kind, query, actor_ref FROM column_sources WHERE workspace_id = $1 AND column_id = $2", [workspaceId, columnId]);
    for (const row of existing.rows) {
      const identity: SourceIdentity = { provider: row.provider, connectionId: row.connection_id, sourceKind: row.source_kind, query: row.query, actorRef: row.actor_ref };
      if (!wanted.has(identityKey(identity))) {
        await client.query("DELETE FROM column_sources WHERE workspace_id = $1 AND id = $2", [workspaceId, row.id]);
      }
    }
    await this.writeSources(client, workspaceId, columnId, type, spec);
  }

  async remove(session: SessionPayload, id: string): Promise<void> {
    await withTenant(session.workspaceId, async (client) => {
      await client.query("SELECT id FROM stream_columns WHERE workspace_id = $1 AND id = $2 AND deleted_at IS NULL FOR UPDATE", [session.workspaceId, id]);
      const result = await client.query(
        "UPDATE stream_columns SET deleted_at = now(), updated_at = now() WHERE workspace_id = $1 AND id = $2 AND deleted_at IS NULL RETURNING id", [session.workspaceId, id]);
      if (result.rows[0] === undefined) throw new NotFoundException(`column ${id} not found`);
      await appendTenantAudit(session.workspaceId, {
        actorUserId: session.userId, actorRole: session.role, action: "columns.deleted",
        targetType: "stream_column", targetId: id,
      });
    });
  }

  /**
   * Explicit order (design §1.4 POST /reorder): the client sends the full
   * ordered id list; positions are rewritten densely. Unknown, foreign or
   * missing ids are a 400 — a partial reorder would corrupt the order.
   */
  async reorder(session: SessionPayload, ids: string[]): Promise<ColumnRecord[]> {
    if (ids.length === 0) return this.list(session.workspaceId);
    if (new Set(ids).size !== ids.length) throw new BadRequestException("reorder ids must be unique");
    return withTenant(session.workspaceId, async (client) => {
      const existing = await client.query<{ id: string }>(
        "SELECT id FROM stream_columns WHERE workspace_id = $1 AND deleted_at IS NULL", [session.workspaceId]);
      if (ids.length !== existing.rows.length || !existing.rows.every((row) => ids.includes(row.id))) {
        throw new BadRequestException("reorder must list every column in the workspace exactly once");
      }
      // Two-phase write: the unique (workspace_id, position) index would fire
      // mid-swap if positions were assigned in place (e.g. moving column 0 to
      // the end collides with the row already at that slot). Parking every row
      // at a high negative offset first leaves the final range untouched, so
      // the second pass can never collide. Offsets are outside the CHECK
      // (position >= 0) only transiently... so instead park at (n + i), which
      // is >= 0, unique within the transaction, and freed before phase two.
      // Phase 1 — park every row at a unique high slot (n, n+1, …) so the
      // unique (workspace_id, position) index can't fire mid-reorder.
      let park = ids.length;
      for (const id of ids) {
        await client.query("UPDATE stream_columns SET position = $3 WHERE workspace_id = $1 AND id = $2", [session.workspaceId, id, park]);
        park += 1;
      }
      // Phase 2 — assign final dense ranks in the REQUESTED order. The parked
      // range is disjoint from [0, n), so no intermediate state collides.
      const byId = new Map(ids.map((id, index) => [id, index] as const));
      const current = await client.query<ColumnRow>(
        `SELECT ${COLUMN_SELECT} FROM stream_columns WHERE workspace_id = $1 AND deleted_at IS NULL`, [session.workspaceId]);
      const records = current.rows
        .map((row) => toRecord(row as ColumnRow))
        .sort((a, b) => (byId.get(a.id) ?? 0) - (byId.get(b.id) ?? 0));
      let rank = 0;
      for (const record of records) {
        await client.query("UPDATE stream_columns SET position = $3, updated_at = now() WHERE workspace_id = $1 AND id = $2", [session.workspaceId, record.id, rank]);
        record.position = rank;
        rank += 1;
      }
      await appendTenantAudit(session.workspaceId, {
        actorUserId: session.userId, actorRole: session.role, action: "columns.reordered",
        targetType: "workspace", targetId: session.workspaceId, diff: { count: ids.length },
      });
      return records;
    });
  }

  /**
   * Lazy seeding (design §6 P1): on the first column-API touch for a workspace
   * with no columns at all, materialize TODAY's implicit /feeds columns — one
   * platform_feed column per feed-capable connection, titled exactly like the
   * implicit column the UI renders today ("Display · handle", kinds ["home"],
   * position by connection creation order), so the Streams page is unchanged.
   * Idempotent by construction: a second call sees a non-empty column list and
   * seeds nothing — including for connections added later.
   */
  async ensureSeeded(workspaceId: string, connections: ConnectionSummary[], capable: (row: ConnectionSummary) => boolean): Promise<{ seeded: number }> {
    const rows = connections.filter(capable);
    if (rows.length === 0) return { seeded: 0 };
    return withTenant(workspaceId, async (client) => {
      // Serialize concurrent first touches without taking a row lock on
      // workspaces itself: a FOR UPDATE here deadlocks with any FK insert
      // into audit_log/column_sources from a nested withTenant client,
      // because FK validation takes FOR KEY SHARE on the parent workspace
      // row. Advisory xact locks never block key-share, so seeding
      // serializes while audits and FK checks proceed.
      await client.query("SELECT pg_advisory_xact_lock(hashtext($1))", [`columns-seed:${workspaceId}`]);
      const existing = await client.query<{ count: string }>(
        "SELECT count(*)::text AS count FROM stream_columns WHERE workspace_id = $1 AND deleted_at IS NULL", [workspaceId]);
      if (Number(existing.rows[0]?.count ?? 0) > 0) return { seeded: 0 };
      const seeder = await client.query<{ user_id: string }>(
        "SELECT m.user_id FROM memberships m JOIN users u ON u.id = m.user_id WHERE m.workspace_id = $1 ORDER BY m.created_at, u.created_at LIMIT 1", [workspaceId]);
      const createdBy = seeder.rows[0]?.user_id;
      if (createdBy === undefined) return { seeded: 0 }; // no member to attribute
      let seeded = 0;
      let position = 0;
      for (const row of rows) {
        const spec = { connectionId: row.id, kinds: ["home"] } as const;
        const inserted = await client.query<{ id: string }>(
          `INSERT INTO stream_columns (workspace_id, type, title, position, spec, refresh_policy, created_by)
           VALUES ($1,'platform_feed',$2,$3,$4,'{}',$5) RETURNING id`,
          [workspaceId, seedTitle(row), position, JSON.stringify(spec), createdBy],
        );
        const columnId = inserted.rows[0]?.id;
        if (columnId === undefined) continue;
        await client.query(
          `INSERT INTO column_sources (workspace_id, column_id, provider, connection_id, source_kind)
           VALUES ($1,$2,$3,$4,'home')
           ON CONFLICT (workspace_id, column_id, provider, source_kind, connection_id, query, actor_ref) DO NOTHING`,
          [workspaceId, columnId, row.provider, row.id],
        );
        seeded += 1;
        position += 1;
      }
      if (seeded > 0) await appendTenantAudit(workspaceId, {
        actorUserId: null, actorRole: "system", action: "columns.seeded",
        targetType: "workspace", targetId: workspaceId, diff: { seeded },
      });
      return { seeded };
    });
  }

  /**
   * import-saved-searches (design §1.4): the localStorage saved-searches
   * contract moves server-side. Each query becomes a saved_search column over
   * the platforms the workspace can actually search, defaulting to every
   * connected feed provider when the client sends none. Idempotent per query:
   * a live saved_search column with the same query is skipped.
   */
  async importSavedSearches(
    session: SessionPayload, queries: string[], platforms: string[],
  ): Promise<{ imported: number; skipped: number }> {
    if (platforms.length === 0) return { imported: 0, skipped: queries.length };
    return withTenant(session.workspaceId, async (client) => {
      let imported = 0;
      let skipped = 0;
      for (const raw of queries) {
        const query = raw.trim();
        if (query === "") { skipped += 1; continue; }
        const duplicate = await client.query<{ id: string }>(
          `SELECT id FROM stream_columns WHERE workspace_id = $1 AND deleted_at IS NULL AND type = 'saved_search'
             AND spec->>'query' = $2 LIMIT 1`, [session.workspaceId, query]);
        if (duplicate.rows[0] !== undefined) { skipped += 1; continue; }
        const spec = { query, platforms, matchMode: "any" } as const;
        const title = clampTitle(query);
        if (title === "") { skipped += 1; continue; }
        const position = await this.nextPosition(client, session.workspaceId);
        const inserted = await client.query<{ id: string }>(
          `INSERT INTO stream_columns (workspace_id, type, title, position, spec, refresh_policy, created_by)
           VALUES ($1,'saved_search',$2,$3,$4,'{}',$5) RETURNING id`,
          [session.workspaceId, title, position, JSON.stringify(spec), session.userId],
        );
        const columnId = inserted.rows[0]?.id;
        if (columnId === undefined) { skipped += 1; continue; }
        await this.writeSources(client, session.workspaceId, columnId, "saved_search",
          parseStreamColumnSpec("saved_search", spec));
        imported += 1;
      }
      if (imported > 0) await appendTenantAudit(session.workspaceId, {
        actorUserId: session.userId, actorRole: session.role, action: "columns.saved_searches_imported",
        targetType: "workspace", targetId: session.workspaceId, diff: { imported, skipped },
      });
      return { imported, skipped };
    });
  }

  /**
   * Poll-time item persistence (§3.3): first writer wins per dedup key (the
   * cross-column dedup), then fan-out rows link the item to every column
   * whose source shares this source's identity — which is how "same account
   * in two columns costs one poll" still surfaces items in both.
   */
  async storeItems(workspaceId: string, sourceId: string, items: NormalizedFeedItem[]): Promise<{ stored: number; fanned: number }> {
    if (items.length === 0) return { stored: 0, fanned: 0 };
    return withTenant(workspaceId, async (client) => {
      const found = await client.query<{ id: string; provider: string; connection_id: string | null; source_kind: ColumnSourceKind; query: string | null; actor_ref: string | null }>(
        "SELECT id, provider, connection_id, source_kind, query, actor_ref FROM column_sources WHERE workspace_id = $1 AND id = $2", [workspaceId, sourceId]);
      const source = found.rows[0];
      if (source === undefined) return { stored: 0, fanned: 0 };
      const sharing = await client.query<{ column_id: string }>(
        `SELECT column_id FROM column_sources
          WHERE workspace_id = $1 AND provider = $2 AND source_kind = $3
            AND connection_id IS NOT DISTINCT FROM $4 AND query IS NOT DISTINCT FROM $5 AND actor_ref IS NOT DISTINCT FROM $6
            AND state <> 'disabled'`,
        [workspaceId, source.provider, source.source_kind, source.connection_id, source.query, source.actor_ref]);
      const columnIds = sharing.rows.map((row) => row.column_id);
      let stored = 0;
      let fanned = 0;
      for (const item of items) {
        const dedupKey = dedupKeyFor(workspaceId, item.provider, item.remoteId);
        // First writer wins per dedup key (the workspace-wide cross-column
        // dedup, §3.3) — the returning row resolves the item id either way.
        const inserted = await client.query<{ id: string }>(
          `INSERT INTO feed_items (workspace_id, dedup_key, provider, remote_id, item_id, author_handle, author_ref, url,
             created_at_remote, fetched_at, content, plain_text, media, metrics, source_id, expires_at)
           VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,now(),$10,$11,$12,$13,$14,$15)
           ON CONFLICT (workspace_id, dedup_key) DO UPDATE SET fetched_at = feed_items.fetched_at RETURNING id`,
          [workspaceId, dedupKey, item.provider, item.remoteId, item.id, item.authorHandle, item.authorRef, item.url,
            item.createdAt.toISOString(), JSON.stringify(item.content), item.plainText, JSON.stringify(item.media),
            JSON.stringify(item.metrics), sourceId, item.expiresAt === null || item.expiresAt === undefined ? null : item.expiresAt.toISOString()],
        );
        const itemId = inserted.rows[0]?.id;
        if (itemId === undefined) continue;
        stored += 1;
        // Fan out regardless of who inserted: an item ingested for one source
        // identity is visible to EVERY column sharing that identity, including
        // items an earlier poll already stored (§3.3) — otherwise the second
        // column's dedup-hit items would never get their link rows.
        for (const columnId of columnIds) {
          const linked = await client.query(
            "INSERT INTO feed_item_columns (workspace_id, item_id, column_id) VALUES ($1,$2,$3) ON CONFLICT DO NOTHING",
            [workspaceId, itemId, columnId]);
          fanned += linked.rowCount ?? 0;
        }
      }
      return { stored, fanned };
    });
  }

  /**
   * Keyset page for GET /feeds/columns/:id/items (§1.4). Cursor rule:
   * (created_at_remote DESC, dedup_key DESC) — a pure position in a total
   * order, so paging survives cache flushes and later insertions alike.
   */
  async columnItems(
    workspaceId: string, columnId: string, options: { cursor?: { at: number; key: string }; limit: number },
  ): Promise<{ items: StoredFeedItem[]; nextCursor: string | null }> {
    return withTenant(workspaceId, async (client) => {
      const column = await client.query("SELECT id FROM stream_columns WHERE workspace_id = $1 AND id = $2 AND deleted_at IS NULL", [workspaceId, columnId]);
      if (column.rows[0] === undefined) throw new NotFoundException(`column ${columnId} not found`);
      const params: unknown[] = [workspaceId, columnId];
      const filter = options.cursor === undefined ? "" : " AND (i.created_at_remote, i.dedup_key) < ($3::timestamptz, $4)";
      if (options.cursor !== undefined) params.push(new Date(options.cursor.at).toISOString(), options.cursor.key);
      params.push(options.limit + 1);
      const result = await client.query(
        `SELECT i.id, i.dedup_key, i.created_at_remote, i.provider, i.remote_id, i.item_id, i.author_handle,
                i.author_ref, i.url, i.content, i.plain_text, i.media, i.metrics, i.expires_at
         FROM feed_items i
         JOIN feed_item_columns fic ON fic.item_id = i.id AND fic.column_id = $2
         WHERE i.workspace_id = $1${filter}
         ORDER BY i.created_at_remote DESC, i.dedup_key DESC
         LIMIT $${params.length}`,
        params,
      );
      const rows = result.rows;
      const hasMore = rows.length > options.limit;
      const items = (hasMore ? rows.slice(0, options.limit) : rows).map((row) => storedItemFromRow(row));
      const last = items[items.length - 1];
      return {
        items,
        nextCursor: hasMore && last
          ? Buffer.from(JSON.stringify({ at: Date.parse(last.createdAtRemote), key: last.dedupKey })).toString("base64url")
          : null,
      };
    });
  }

  private async nextPosition(client: PoolClient, workspaceId: string): Promise<number> {
    const result = await client.query<{ max: number | null }>(
      "SELECT max(position) AS max FROM stream_columns WHERE workspace_id = $1 AND deleted_at IS NULL", [workspaceId]);
    return (result.rows[0]?.max ?? -1) + 1;
  }

  /** A spec's references must exist in the workspace before the column can persist. */
  private async assertSpecReferenceable(client: PoolClient, workspaceId: string, type: StreamColumnType, spec: StreamColumnSpec): Promise<void> {
    if (type === "platform_feed") {
      const connectionId = (spec as import("@tactical-media-broadcast-hub/core").PlatformFeedSpec).connectionId;
      const result = await client.query(
        "SELECT 1 FROM connections WHERE workspace_id = $1 AND id = $2 AND disconnected_at IS NULL", [workspaceId, connectionId]);
      if (result.rows[0] === undefined) throw new UnprocessableEntityException("platform_feed: connectionId does not reference a connection in this workspace");
      return;
    }
    const providers = columnProviders(type, spec);
    if (providers.length === 0) return;
    const result = await client.query<{ provider: string }>(
      "SELECT DISTINCT provider FROM connections WHERE workspace_id = $1 AND disconnected_at IS NULL AND provider = ANY($2)", [workspaceId, providers]);
    const present = new Set(result.rows.map((row) => row.provider));
    const missing = [...new Set(providers)].filter((provider) => !present.has(provider));
    if (missing.length > 0) throw new UnprocessableEntityException(`no connection for provider(s): ${missing.join(", ")}`);
  }

  private async read(client: PoolClient, workspaceId: string, id: string): Promise<ColumnRecord> {
    const result = await client.query(
      `SELECT ${COLUMN_SELECT} FROM stream_columns WHERE workspace_id = $1 AND id = $2 AND deleted_at IS NULL`, [workspaceId, id]);
    const row = result.rows[0];
    if (row === undefined) throw new NotFoundException(`column ${id} not found`);
    return toRecord(row as ColumnRow);
  }
}

interface FeedItemRow {
  id: string; dedup_key: string; created_at_remote: Date; provider: string; remote_id: string; item_id: string;
  author_handle: string; author_ref: string | null; url: string | null; content: unknown; plain_text: string;
  media: unknown; metrics: unknown; expires_at: Date | null;
}

function storedItemFromRow(row: FeedItemRow): StoredFeedItem {
  const metrics = (typeof row.metrics === "object" && row.metrics !== null ? row.metrics : {}) as Record<string, number>;
  const item: NormalizedFeedItem = {
    id: row.item_id,
    remoteId: row.remote_id,
    provider: row.provider,
    // The store page does not carry a live connection id; the column's source
    // rows hold it, so the connectionId field is filled by the caller when it
    // matters (the items endpoint derives it from the column spec).
    connectionId: "",
    authorHandle: row.author_handle,
    authorRef: row.author_ref,
    url: row.url,
    text: row.plain_text,
    content: row.content as NormalizedFeedItem["content"],
    plainText: row.plain_text,
    createdAt: row.created_at_remote,
    mediaUrls: [],
    media: Array.isArray(row.media) ? row.media as NormalizedFeedItem["media"] : [],
    metrics,
    expiresAt: row.expires_at,
  };
  return { id: row.id, dedupKey: row.dedup_key, createdAtRemote: new Date(row.created_at_remote).toISOString(), item };
}