T-HUB/packages/core/src/stream-columns.ts

/**
 * Streams columns (WP3 §1.2): typed `spec` per column type + clamped
 * `refresh_policy`. Invalid specs are rejected before insert (the API maps
 * StreamColumnsValidationError → 422) and never stored; clamps are applied,
 * not errors — intervalMs floors to 60 s / caps at 1 h, titles trim+cap at 80
 * chars, positions stay sane dense ranks.
 *
 * Dependency-free by house rule (see chat-contract.schema.ts): plain parse
 * functions, no zod. The four persisted types and their shapes follow design
 * §1.1/§1.2; `saved_search` shares the keyword engine, it only differs in the
 * column `type` label.
 */
import type { ProviderId } from "./connection.js";

export type StreamColumnType = "platform_feed" | "keyword" | "account_list" | "saved_search";

export const STREAM_COLUMN_TYPES: readonly StreamColumnType[] = [
  "platform_feed",
  "keyword",
  "account_list",
  "saved_search",
] as const;

export type PlatformFeedKind = "home" | "mentions";

/** `{ connectionId, kinds }` — kinds defaults to ["home"] (design §1.2). */
export interface PlatformFeedSpec {
  connectionId: string;
  kinds: PlatformFeedKind[];
}

/** Shared by `keyword` and `saved_search` (§1.1: saved_search is keyword). */
export interface KeywordSpec {
  /** 1..200 chars after trimming; no NUL/control chars. */
  query: string;
  /** Provider ids, e.g. ["bluesky","mastodon","youtube"]; deduped. */
  platforms: ProviderId[];
  /** Reserved; v1 = single term set, no boolean ops. */
  matchMode: "any";
}

export interface AccountListEntry {
  provider: ProviderId;
  /** Free text: "@user@instance.social" | "@user" | "UC…" (YT channel id). */
  handle: string;
}

/** 1..50 entries (design §1.2). */
export interface AccountListSpec {
  accounts: AccountListEntry[];
}

export type StreamColumnSpec = PlatformFeedSpec | KeywordSpec | AccountListSpec;

/** `{ intervalMs?, paused? }` — clamped server-side (design §1.2). */
export interface RefreshPolicy {
  /** 60_000..3_600_000 after clamping. */
  intervalMs?: number;
  paused?: boolean;
}

export const QUERY_MAX = 200;
export const ACCOUNTS_MAX = 50;
export const PLATFORMS_MAX = 200;
/** Title cap: matches the stream_columns.title CHECK in migration 021. */
export const STREAM_COLUMN_TITLE_MAX = 80;
export const STREAM_COLUMN_POSITION_DEFAULT = 0;
const POSITION_MAX = 2_147_483_647; // integer column

const INTERVAL_FLOOR = 60_000;
const INTERVAL_CEIL = 3_600_000;

/** Default cadence per column type (design §1.2). */
export const DEFAULT_INTERVAL_MS: Record<StreamColumnType, number> = {
  platform_feed: 60_000,
  keyword: 120_000,
  account_list: 180_000,
  saved_search: 300_000,
};

/** API layer maps this → 422; never stored on failure. */
export class StreamColumnsValidationError extends Error {
  constructor(message: string) {
    super(message);
    this.name = "StreamColumnsValidationError";
  }
}

function fail(message: string): never {
  throw new StreamColumnsValidationError(message);
}

function asObject(value: unknown, what: string): Record<string, unknown> {
  if (typeof value !== "object" || value === null || Array.isArray(value)) {
    return fail(`${what} must be a JSON object`);
  }
  return value as Record<string, unknown>;
}

function isControlCharFree(value: string): boolean {
  return !/[\u0000-\u001f\u007f]/.test(value);
}

function providerId(value: unknown, what: string): ProviderId {
  if (typeof value !== "string" || value.length === 0 || value.length > 64) {
    return fail(`${what} must be a 1..64-char provider id`);
  }
  if (!isControlCharFree(value)) return fail(`${what} must not contain control characters`);
  return value;
}

// ---------------------------------------------------------------------------
// spec: platform_feed
// ---------------------------------------------------------------------------

function parsePlatformFeedSpec(value: unknown): PlatformFeedSpec {
  const spec = asObject(value, "platform_feed spec");
  if (typeof spec["connectionId"] !== "string" || spec["connectionId"].length === 0) {
    fail("platform_feed: connectionId must be a non-empty string");
  }
  if (spec["connectionId"].length > 200) {
    fail("platform_feed: connectionId exceeds 200 chars");
  }
  const rawKinds = spec["kinds"];
  const kinds: PlatformFeedKind[] =
    rawKinds === undefined ? ["home"] : parseKinds(rawKinds);
  return { connectionId: spec["connectionId"], kinds };
}

function parseKinds(value: unknown): PlatformFeedKind[] {
  if (!Array.isArray(value)) {
    fail("platform_feed: kinds must be a non-empty array of \"home\"|\"mentions\"");
  }
  const kinds: PlatformFeedKind[] = [];
  for (const kind of value) {
    if (kind !== "home" && kind !== "mentions") {
      fail("platform_feed: kinds may contain only \"home\"|\"mentions\"");
    }
    if (!kinds.includes(kind)) kinds.push(kind);
  }
  if (kinds.length === 0) {
    fail("platform_feed: kinds must be a non-empty array of \"home\"|\"mentions\"");
  }
  return kinds;
}

// ---------------------------------------------------------------------------
// spec: keyword / saved_search — one engine, two labels (design §1.1)
// ---------------------------------------------------------------------------

function parseKeywordSpec(value: unknown): KeywordSpec {
  const spec = asObject(value, "keyword spec");
  const rawQuery = spec["query"];
  if (typeof rawQuery !== "string") fail("keyword: query must be a string");
  const query = rawQuery.trim();
  if (query.length < 1) fail("keyword: query must be 1..200 chars after trimming");
  if (query.length > QUERY_MAX) fail(`keyword: query exceeds ${QUERY_MAX} chars after trimming`);
  if (!isControlCharFree(query)) {
    fail("keyword: query must not contain NUL/control characters");
  }
  const platforms = parsePlatforms(spec["platforms"]);
  const matchMode = spec["matchMode"];
  if (matchMode !== undefined && matchMode !== "any") {
    fail("keyword: matchMode must be \"any\" (no boolean operators in v1)");
  }
  return { query, platforms, matchMode: "any" };
}

function parsePlatforms(value: unknown): ProviderId[] {
  if (!Array.isArray(value) || value.length < 1 || value.length > PLATFORMS_MAX) {
    fail(`keyword: platforms must be an array of 1..${PLATFORMS_MAX} provider ids`);
  }
  const platforms: ProviderId[] = [];
  for (const platform of value) {
    const id = providerId(platform, "keyword: platform");
    if (!platforms.includes(id)) platforms.push(id); // dedupe silently
  }
  return platforms;
}

// ---------------------------------------------------------------------------
// spec: account_list
// ---------------------------------------------------------------------------

function parseAccountListSpec(value: unknown): AccountListSpec {
  const spec = asObject(value, "account_list spec");
  const rawAccounts = spec["accounts"];
  if (!Array.isArray(rawAccounts) || rawAccounts.length < 1 || rawAccounts.length > ACCOUNTS_MAX) {
    fail(`account_list: accounts must be an array of 1..${ACCOUNTS_MAX} entries`);
  }
  const accounts: AccountListEntry[] = [];
  for (const entry of rawAccounts) {
    const record = asObject(entry, "account_list: account");
    const provider = providerId(record["provider"], "account_list: provider");
    const handle = record["handle"];
    if (typeof handle !== "string") fail("account_list: handle must be a string");
    const trimmed = handle.trim();
    if (trimmed.length < 1 || trimmed.length > 200) {
      fail("account_list: handle must be 1..200 chars after trimming");
    }
    if (!isControlCharFree(trimmed)) {
      fail("account_list: handle must not contain NUL/control characters");
    }
    accounts.push({ provider, handle: trimmed });
  }
  return { accounts };
}

// ---------------------------------------------------------------------------
// refresh_policy
// ---------------------------------------------------------------------------

function parseRefreshPolicyObject(value: unknown): RefreshPolicy {
  const result: RefreshPolicy = {};
  if (value === undefined || value === null) return result; // column default '{}'
  const policy = asObject(value, "refresh_policy");
  if (policy["intervalMs"] !== undefined) {
    const intervalMs = policy["intervalMs"];
    if (typeof intervalMs !== "number") {
      fail("refresh_policy: intervalMs must be a number");
    }
    result.intervalMs = clampIntervalMs(intervalMs); // non-finite -> floor
  }
  if (policy["paused"] !== undefined) {
    if (typeof policy["paused"] !== "boolean") {
      fail("refresh_policy: paused must be a boolean");
    }
    result.paused = policy["paused"];
  }
  return result;
}

/** Clamp to the design §1.2 window 60_000..3_600_000 (rounded, finite). */
export function clampIntervalMs(intervalMs: number): number {
  if (Number.isNaN(intervalMs)) return INTERVAL_FLOOR;
  return Math.min(Math.max(Math.round(intervalMs), INTERVAL_FLOOR), INTERVAL_CEIL);
}

/** Effective cadence when the user set none (design §1.2 defaults). */
export function defaultIntervalMs(type: StreamColumnType): number {
  return DEFAULT_INTERVAL_MS[type];
}

// ---------------------------------------------------------------------------
// Public entry points
// ---------------------------------------------------------------------------

/** Maps a column type to its exact persisted spec shape (design §1.2). */
export type SpecForType<T extends StreamColumnType> =
  T extends "platform_feed" ? PlatformFeedSpec
  : T extends "account_list" ? AccountListSpec
  : KeywordSpec; // keyword + saved_search share the keyword engine (§1.1)

/**
 * Validate + clamp a `spec` for a column type. Throws
 * StreamColumnsValidationError (→ 422 at the API boundary); returns the
 * canonical clamped form to persist as JSONB, typed per column type.
 */
export function parseStreamColumnSpec<T extends StreamColumnType>(
  type: T,
  value: unknown,
): SpecForType<T> {
  switch (type) {
    case "platform_feed":
      return parsePlatformFeedSpec(value) as SpecForType<T>;
    case "keyword":
    case "saved_search":
      return parseKeywordSpec(value) as SpecForType<T>;
    case "account_list":
      return parseAccountListSpec(value) as SpecForType<T>;
  }
}

/**
 * Validate + clamp `refresh_policy` (default `{}`). intervalMs clamps to
 * 60_000..3_600_000. The budget-derived floor ("never below the
 * per-connection budget interval") is applied by the scheduler on top of
 * this; this layer enforces only the global clamp window.
 */
export function parseRefreshPolicyClamped(value: unknown = {}): RefreshPolicy {
  return parseRefreshPolicyObject(value);
}

// ---------------------------------------------------------------------------
// Small shared clamps
// ---------------------------------------------------------------------------

export function isKeywordLike(type: StreamColumnType): type is "keyword" | "saved_search" {
  return type === "keyword" || type === "saved_search";
}

/** Trim + cap at 80 chars so a persist can never trip the DB CHECK. */
export function clampTitle(title: string): string {
  const trimmed = title.trim();
  return trimmed.length > STREAM_COLUMN_TITLE_MAX
    ? trimmed.slice(0, STREAM_COLUMN_TITLE_MAX)
    : trimmed;
}

/** Dense-rank positions: finite, non-negative integers. */
export function clampPosition(position: number): number {
  if (Number.isNaN(position)) return STREAM_COLUMN_POSITION_DEFAULT;
  return Math.min(Math.max(Math.round(position), STREAM_COLUMN_POSITION_DEFAULT), POSITION_MAX);
}