linkedin-studio/scripts/brain/src/ingest.ts

/**
 * Ingest — the published-post gold signal (SB-S1).
 *
 * Captures the user's ACTUAL published posts into `ingest/published/` tagged
 * `provenance=published`, so the voice/profile-learning surface can learn from
 * human-published content ONLY (the model-collapse guard — never from ai-draft).
 *
 * This module has two halves:
 *  - PURE: `PublishedRecord` + `parsePublishedRecord` / `serializePublishedRecord`
 *    (a file-per-post constrained-header grammar; NO YAML dep, mirroring the brain
 *    profile.md line-grammar idiom). `parse ∘ serialize = identity` (SC2).
 *  - IO: `writePublished` / `ingestText` / `scanInbox` / `listPublished` — idempotent,
 *    collision-safe, create-on-demand under the brain `dataRoot` (SC1/SC3/SC4).
 *
 * The record id is `mintContentId(VERBATIM body)` (id.ts) — byte-identity, so two
 * structurally-different posts never collide and the write path never silently
 * drops a differing body.
 */

import { existsSync, mkdirSync, readFileSync, readdirSync, writeFileSync } from "node:fs";
import { join } from "node:path";

import { dataRoot } from "./dataRoot.js";
import { mintContentId, normalizeProvenance } from "./id.js";

/** One ingested published post. `provenance` is always `published` in SB-S1. */
export interface PublishedRecord {
  /** mintContentId(body) — 12 hex, the dedupe key + filename stem. */
  id: string;
  /** Always `published` — the type pins it; parse rejects any other value. */
  provenance: "published";
  /** YYYY-MM-DD — when the post was published (operator-supplied or = captured_at). */
  published_date: string;
  /** YYYY-MM-DD — when this record was ingested. */
  captured_at: string;
  /** `manual` (default) | a future connector token. */
  source: string;
  /**
   * SB-S3c — the raw material this post was built from: specifics-bank + trends
   * ids (12-hex), threaded hub-side so the post↔specific↔trend↔analytics graph is
   * assemblable. Default `[]`; serialized only when non-empty (omit-empty keeps a
   * pre-S3c record byte-identical). Validated to 12-hex on parse + at the producer.
   */
  specifics: string[];
  trends: string[];
  /** The verbatim post body — byte-exact, may contain anything (`]`/`|`/newlines/`---`). */
  body: string;
}

/** The one separator string, used identically by serialize and parse. */
const SENTINEL = "\n---\n";

/**
 * Serialize a record to the file grammar: a fixed 5-line header, the `---`
 * sentinel, then the verbatim body. No title, no blank-line padding (the body is
 * captured byte-exact, so any padding would break the round-trip identity).
 */
export function serializePublishedRecord(rec: PublishedRecord): string {
  const header = [
    `id: ${rec.id}`,
    `provenance: ${rec.provenance}`,
    `published_date: ${rec.published_date}`,
    `captured_at: ${rec.captured_at}`,
    `source: ${rec.source}`,
  ];
  // SB-S3c: the raw-material id lines are appended AFTER `source:` and emitted ONLY
  // when non-empty — so a pre-S3c record (empty arrays) serializes to the unchanged
  // 5-line header, byte-identical (SC2). Body-only `mintContentId` is unaffected.
  if (rec.specifics.length > 0) header.push(`specifics: ${rec.specifics.join(",")}`);
  if (rec.trends.length > 0) header.push(`trends: ${rec.trends.join(",")}`);
  return header.join("\n") + SENTINEL + rec.body;
}

const ID_RE = /^[0-9a-f]{12}$/;

/**
 * SB-S3c: read an optional comma-separated 12-hex id list from the header slice.
 * Absent key → `[]` (a NON-throwing reader — NOT `headerScalar`, which throws on a
 * missing key and would break every pre-S3c record). A present-but-malformed id
 * throws (never silently dropped).
 */
function headerIdList(header: string, key: string): string[] {
  const m = header.match(new RegExp(`^${key}:\\s*(.*?)\\s*$`, "m"));
  if (!m) return [];
  return m[1]
    .split(",")
    .map((s) => s.trim())
    .filter((s) => s !== "")
    .map((id) => {
      if (!ID_RE.test(id)) {
        throw new Error(`malformed published record: bad ${key} id ${JSON.stringify(id)}`);
      }
      return id;
    });
}

function headerScalar(header: string, key: string): string {
  const m = header.match(new RegExp(`^${key}:\\s*(.*?)\\s*$`, "m"));
  if (!m) throw new Error(`malformed published record: missing "${key}:" header`);
  return m[1];
}

/**
 * Parse a record back from the file grammar. Splits on the FIRST `---` sentinel
 * (the header lines are constrained `key: value` pairs that never contain one), so
 * a body that itself contains `\n---\n` is preserved verbatim. The header scalars
 * are read only from the header slice — a header-shaped body line cannot leak in.
 * `provenance` is asserted to be `published`: a `published/` record carrying any
 * other provenance is a corruption signal and throws (never silently accepted).
 */
export function parsePublishedRecord(text: string): PublishedRecord {
  const idx = text.indexOf(SENTINEL);
  if (idx === -1) throw new Error("malformed published record: no `---` separator");
  const header = text.slice(0, idx);
  const body = text.slice(idx + SENTINEL.length);

  const id = headerScalar(header, "id");
  if (!/^[0-9a-f]{12}$/.test(id)) {
    throw new Error(`malformed published record: bad id ${JSON.stringify(id)}`);
  }
  const provenance = normalizeProvenance(headerScalar(header, "provenance"));
  if (provenance !== "published") {
    throw new Error(
      `corrupt published record: provenance is "${provenance}", expected "published"`,
    );
  }
  return {
    id,
    provenance,
    published_date: headerScalar(header, "published_date"),
    captured_at: headerScalar(header, "captured_at"),
    source: headerScalar(header, "source"),
    specifics: headerIdList(header, "specifics"),
    trends: headerIdList(header, "trends"),
    body,
  };
}

// ── IO ───────────────────────────────────────────────────────────────────────
//
// All IO resolves through the brain package's own `dataRoot` (LINKEDIN_STUDIO_DATA
// overrides the root for tests + installs). Create-on-demand: `ingest/published/`
// is mkdir'd as needed, so ingest does NOT depend on a prior `brain init`.

const PUBLISHED_SUB = join("ingest", "published");
const INBOX_SUB = join("ingest", "inbox");

/** The body stored at `path`, or `null` if it can't be read/parsed (treat as differing). */
function storedBody(path: string): string | null {
  try {
    return parsePublishedRecord(readFileSync(path, "utf8")).body;
  } catch {
    return null;
  }
}

export interface WriteResult {
  /** true iff a file was written (false = identical record already present). */
  written: boolean;
  /** The path written or matched. */
  path: string;
  /** true iff an id-collision with a DIFFERING body forced a disambiguated name. */
  collision?: boolean;
}

/**
 * Write a record idempotently + collision-safely. If `<id>.md` already holds the
 * SAME body → no-op (`written:false`). If it holds a DIFFERENT body (a 48-bit hash
 * collision or an unreadable file) → write the next free `<id>-N.md` rather than
 * clobber or silently skip — so a differing gold record is never lost (B2).
 */
export function writePublished(rec: PublishedRecord): WriteResult {
  const dir = dataRoot(PUBLISHED_SUB);
  mkdirSync(dir, { recursive: true });
  const text = serializePublishedRecord(rec);

  const primary = join(dir, `${rec.id}.md`);
  if (!existsSync(primary)) {
    writeFileSync(primary, text, "utf8");
    return { written: true, path: primary };
  }
  if (storedBody(primary) === rec.body) return { written: false, path: primary };

  for (let n = 2; ; n++) {
    const cand = join(dir, `${rec.id}-${n}.md`);
    if (!existsSync(cand)) {
      writeFileSync(cand, text, "utf8");
      return { written: true, path: cand, collision: true };
    }
    if (storedBody(cand) === rec.body) return { written: false, path: cand, collision: true };
  }
}

export interface IngestResult extends WriteResult {
  record: PublishedRecord;
}

/**
 * Mint a record from a verbatim post body and write it. `provenance` is always
 * `published`; `published_date` defaults to `captured_at` when not supplied. Pure
 * inputs (the caller supplies `captured_at`), idempotent + collision-safe write.
 */
export function ingestText(opts: {
  body: string;
  captured_at: string;
  source?: string;
  published_date?: string;
  specifics?: string[];
  trends?: string[];
}): IngestResult {
  // SB-S3c producer guard: a malformed raw-material id fails fast HERE (not only on
  // a later re-parse). Empty/absent arrays skip validation, so every pre-S3c caller
  // (incl. scanInbox) is unaffected.
  const validateIds = (ids: string[] | undefined, kind: string): string[] => {
    const list = ids ?? [];
    for (const id of list) {
      if (!ID_RE.test(id)) throw new Error(`ingest: bad ${kind} id ${JSON.stringify(id)}`);
    }
    return list;
  };
  const record: PublishedRecord = {
    id: mintContentId(opts.body),
    provenance: "published",
    published_date: opts.published_date ?? opts.captured_at,
    captured_at: opts.captured_at,
    source: opts.source ?? "manual",
    specifics: validateIds(opts.specifics, "specifics"),
    trends: validateIds(opts.trends, "trends"),
    body: opts.body,
  };
  return { record, ...writePublished(record) };
}

export interface ScanResult {
  /** ids of records written this scan. */
  processed: string[];
  /** ids of records already present (skipped as duplicates). */
  skipped: string[];
}

/**
 * Process the inbox drop-zone into `published/`. Reads TOP-LEVEL `*.md` files only,
 * skipping dotfiles (`.DS_Store` etc.) and non-markdown — no recursion. An empty or
 * absent inbox is a clean no-op. Non-destructive in SB-S1: processed inbox files are
 * left in place (a re-scan re-skips via the published dedup).
 */
export function scanInbox(opts: { captured_at: string; source?: string }): ScanResult {
  const dir = dataRoot(INBOX_SUB);
  const processed: string[] = [];
  const skipped: string[] = [];
  if (!existsSync(dir)) return { processed, skipped };

  for (const entry of readdirSync(dir, { withFileTypes: true })) {
    if (!entry.isFile() || entry.name.startsWith(".") || !entry.name.endsWith(".md")) continue;
    const body = readFileSync(join(dir, entry.name), "utf8");
    const res = ingestText({ body, captured_at: opts.captured_at, source: opts.source });
    (res.written ? processed : skipped).push(res.record.id);
  }
  return { processed, skipped };
}

export interface PublishedSummary {
  id: string;
  provenance: string;
  published_date: string;
  firstLine: string;
}

/**
 * List the published gold corpus. Each file is parsed inside a try/catch — a
 * malformed/hand-dropped file is counted as `skipped`, never crashing the command.
 * Surfaces `provenance` so the operator can eyeball that nothing `ai-draft` leaked.
 */
export function listPublished(): { records: PublishedSummary[]; skipped: number } {
  const dir = dataRoot(PUBLISHED_SUB);
  const records: PublishedSummary[] = [];
  let skipped = 0;
  if (!existsSync(dir)) return { records, skipped };

  for (const name of readdirSync(dir)) {
    if (name.startsWith(".") || !name.endsWith(".md")) continue;
    try {
      const rec = parsePublishedRecord(readFileSync(join(dir, name), "utf8"));
      records.push({
        id: rec.id,
        provenance: rec.provenance,
        published_date: rec.published_date,
        firstLine: rec.body.split("\n", 1)[0],
      });
    } catch {
      skipped++;
    }
  }
  records.sort((a, b) => a.published_date.localeCompare(b.published_date));
  return { records, skipped };
}