T-HUB/apps/api/src/intake/intake-ingest.ts

/**
 * The intake ingest funnel: raw adapter items → P0c normalize → content-hash
 * → idempotent upsert into intake_items (027).
 *
 * Every item passes `normalizeFeedItem` (task requirement 3), so `content`
 * lands as the closed safe-content AST — never raw provider HTML — and media
 * as proxy-minted refs. `text` (the FTS corpus) is derived from the AST via
 * `astToPlainText`, never the raw provider string.
 *
 * `content_hash` is the schema's dedup key — UNIQUE (workspace_id,
 * content_hash): sha256 of the NFKC + whitespace-collapsed AST plain text
 * plus the sorted media origin identity. The same statement arriving via RSS
 * and a Mastodon boost (or the same URL watched twice) hashes equal and
 * stores once; first writer wins, later deliveries no-op.
 */
import { createHash } from "node:crypto";
import type { PoolClient } from "pg";
import { normalizeFeedItem, astToPlainText, type MediaRef, type NormalizedFeedItem } from "../feeds/feed-normalize.js";
import { connectionFor, type RawIntakeItem } from "./adapters/adapter.js";
import { PROVIDER_BY_KIND, type IntakeKind } from "./intake-kinds.js";
import type { IntakeParams } from "./intake-params.js";
import type { HarvestedMedia } from "./media-harvest.js";

export type IntakeMediaRef = MediaRef & {
  /** S3 content-addressed key when the harvest stored the origin file. */
  storageKey?: string;
  hash?: string;
  sizeBytes?: number;
  contentType?: string;
  harvestStatus?: "stored" | "skipped" | "too_large" | "failed";
  originUrl?: string;
};

export interface NormalizeOutcome {
  normalized: NormalizedFeedItem | null;
  plainText: string;
  contentHash: string;
}

/** NFKC + whitespace-collapse, per the 027 content_hash note. */
export function collapseText(plainText: string): string {
  return plainText.normalize("NFKC").replace(/\s+/g, " ").trim();
}

/** sha256(collapsed plain text + sorted unique media origin URLs). */
export function itemContentHash(plainText: string, mediaUrls: readonly string[]): string {
  const origins = [...new Set(mediaUrls.filter((url) => typeof url === "string" && url !== ""))].sort();
  return createHash("sha256")
    .update(JSON.stringify([collapseText(plainText), origins]))
    .digest("hex");
}

/** Normalize one raw item; normalized = null when the funnel's validation drops it. */
export function normalizeRawItem(raw: RawIntakeItem, kind: IntakeKind, params: IntakeParams["params"],
  workspaceId: string, sourceId: string): NormalizeOutcome {
  const connection = connectionFor(kind, params, sourceId, workspaceId);
  const normalized = normalizeFeedItem(raw, connection, raw.normalize);
  if (normalized === null) return { normalized: null, plainText: "", contentHash: "" };
  const plainText = astToPlainText(normalized.content);
  return { normalized, plainText, contentHash: itemContentHash(plainText, raw.mediaUrls) };
}

/** Merge the P0c proxied media refs with the harvest outcome per origin URL. */
export function mergeMediaRefs(normalized: NormalizedFeedItem, harvested: HarvestedMedia[]): IntakeMediaRef[] {
  const byUrl = new Map(harvested.map((entry) => [entry.url, entry]));
  return normalized.media.map((ref): IntakeMediaRef => {
    const origin = originOfProxyUrl(ref.proxyUrl);
    const hit = origin === null ? undefined : byUrl.get(origin);
    if (!hit) return ref;
    return {
      ...ref,
      originUrl: hit.url,
      ...(hit.status === "stored" ? {
        storageKey: hit.storageKey,
        hash: hit.hash,
        sizeBytes: hit.sizeBytes,
        contentType: hit.contentType,
        harvestStatus: "stored" as const,
      } : { harvestStatus: hit.status }),
    };
  });
}

/** Extracts the origin URL back out of a media-proxy URL. */
export function originOfProxyUrl(proxyUrl: string): string | null {
  const marker = "url=";
  const at = proxyUrl.indexOf(marker);
  if (at < 0) return null;
  try { return decodeURIComponent(proxyUrl.slice(at + marker.length)); } catch { return null; }
}

export interface UpsertResult {
  inserted: number;
  unchanged: number;
}

/**
 * Idempotent upsert inside the caller's tenant transaction. The conflict
 * target is the content-addressed key — (workspace_id, content_hash) — so a
 * re-poll that yields the same content writes nothing, and cross-source
 * duplicates (RSS + Mastodon boost of the same statement) collapse too.
 */
export async function upsertItems(client: PoolClient, workspaceId: string, sourceId: string, kind: IntakeKind,
  rows: Array<{ raw: RawIntakeItem; outcome: NormalizeOutcome; harvested: HarvestedMedia[] }>): Promise<UpsertResult> {
  const result: UpsertResult = { inserted: 0, unchanged: 0 };
  for (const entry of rows) {
    const { normalized, plainText, contentHash } = entry.outcome;
    if (normalized === null) continue;
    const mediaRefs = mergeMediaRefs(normalized, entry.harvested);
    const attribution: Record<string, unknown> = { ...entry.raw.attribution, provider: PROVIDER_BY_KIND[kind] };
    const inserted = await client.query(
      `INSERT INTO intake_items (workspace_id, source_id, remote_ref, content_hash, published_at,
         text, content, media_refs, attribution)
       VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9)
       ON CONFLICT (workspace_id, content_hash) DO NOTHING`,
      [workspaceId, sourceId, entry.raw.remoteRef, contentHash, normalized.createdAt.toISOString(),
        plainText, JSON.stringify(normalized.content), JSON.stringify(mediaRefs), JSON.stringify(attribution)]);
    if ((inserted.rowCount ?? 0) > 0) result.inserted += 1;
    else result.unchanged += 1;
  }
  return result;
}