<?php
/**
 * social-ingest.php — FB/IG customer-message ingest for Lineup (SEPARATE from SMS/comms per Ray's ruling).
 *
 * POST /social_2026/social-ingest.php   (X-API-Key: <fleet key>)
 *   sorts a message into a new or existing conversation on (platform, conversation_id),
 *   dedupes at the ENGINE (id-bearing rows by external_id; id-less by generated composite),
 *   caches avatar + attachments at ingest, then notifies the Lineup UI the same way SMS does.
 * GET  ?action=health   — counts, no auth needed for the shape check
 * GET  ?action=conversations | ?action=messages&conversation_id=… — read (auth)
 *
 * Store: social_2026 (isolated schema, base tables, no prod views). Built 2026-09-27, Zora.
 * CORS is at the Apache vhost level for data.finestoflines.net — NOT set here.
 */
header('Content-Type: application/json; charset=utf-8');
if (($_SERVER['REQUEST_METHOD'] ?? '') === 'OPTIONS') { http_response_code(200); exit; }

require_once '/var/www/data.finestoflines.net/config/db-config.php';
$DBc = $DB_CONFIG['production'];
try {
    $pdo = new PDO("mysql:host={$DBc['host']};dbname=social_2026;charset=utf8mb4", $DBc['user'], $DBc['pass'],
        [PDO::ATTR_ERRMODE => PDO::ERRMODE_EXCEPTION, PDO::ATTR_DEFAULT_FETCH_MODE => PDO::FETCH_ASSOC]);
} catch (Throwable $e) { http_response_code(500); echo json_encode(['status'=>'error','error'=>'db']); exit; }

$action = $_GET['action'] ?? '';
if ($action === 'health') { // shape check, no auth — never returns data rows
    $c = fn($t) => (int)$pdo->query("SELECT COUNT(*) FROM $t")->fetchColumn();
    echo json_encode(['status'=>'ok','store'=>'social_2026',
        'counts'=>['conversations'=>$c('conversations'),'messages'=>$c('messages'),'attachments'=>$c('attachments')]]);
    exit;
}

// --- Auth: fleet X-API-Key (same shape as fb-ingest) ---
$FLEET_KEYS = require '/var/www/data.finestoflines.net/config/social-keys.php';
$key = $_SERVER['HTTP_X_API_KEY'] ?? '';
$caller = $FLEET_KEYS[$key] ?? null;

// ============================================================================================
// BATCH 2 (send/sync lane, 2026-09-29). ⛔ AUTH IS SELECTED BY *ACTION* HERE, NOT BY METHOD.
// The original model below is method-based: POST => fleet key, GET => per-person Citadel token.
// The send lane needs BOTH inversions of that, so these actions are dispatched FIRST, each with
// its auth stated explicitly and enforced before any work:
//   FLEET-KEY actions  (Pulse: server-side, shared key, NO Citadel token exists for her)
//       pending-sends · send-result · outbound-media
//   CITADEL actions    (staff, from the browser, per-person token — Solace ruling (a) 2026-09-27)
//       send · upload · sync-request      [sync-request lands with the sync_requests table]
// ⛔ FAIL-CLOSED BY CONSTRUCTION: an action not named in either list falls through to the ORIGINAL
// method-based rules untouched, so nothing is loosened by omission and the ingest path is unchanged.
// ⛔ The two lists are disjoint and neither accepts the other's credential. A fleet key must never
// reach a per-person read, and a staff token must never reach Pulse's queue.
// ============================================================================================
$FLEET_ACTIONS = ['pending-sends','send-result','outbound-media'];
$STAFF_ACTIONS = ['send','upload'];

if (in_array($action, $FLEET_ACTIONS, true)) {
    if ($caller === null) { http_response_code(401); echo json_encode(['status'=>'error','error'=>'auth']); exit; }

    // --- outbound-media: Pulse fetches the BYTES of an attachment she is about to send ---
    // ⛔ AUTHORIZATION IS A DATABASE QUESTION, NOT A FILENAME PATTERN. Accepting the fleet key for
    // "att_* only" would NOT mean outbound: every INBOUND client photo is att_-prefixed too, so that
    // shape would hand any holder of the SHARED key every client attachment ever cached. Instead the
    // file must be attached to a message in THIS conversation that Pulse still has work on.
    // Inbound media is unreachable BY CONSTRUCTION (those rows are status='received'), and the grant
    // EXPIRES ITSELF as rows settle — no TTL to manage, no signing secret to rotate.
    if ($action === 'outbound-media') {
        $f = basename((string)($_GET['file'] ?? ''));
        if (!preg_match('/^(avatar|att)_[0-9a-f]{16}$/', $f)) {
            http_response_code(400); echo json_encode(['status'=>'error','error'=>'bad file name']); exit;
        }
        // queued|failed = still her work. 'sent' gets a 15-MINUTE GRACE so a retry can re-fetch.
        $q = $pdo->prepare(
            "SELECT 1 FROM attachments a
               JOIN messages m ON m.id = a.message_pk
              WHERE a.cached_filename = ?
                AND m.direction = 'out'
                AND ( m.status IN ('queued','failed')
                   OR (m.status = 'sent' AND m.created_at > (NOW() - INTERVAL 15 MINUTE)) )
              LIMIT 1");
        $q->execute([$f]);
        if ($q->fetchColumn() === false) {
            http_response_code(404); echo json_encode(['status'=>'error','error'=>'not an outbound attachment you may fetch']); exit;
        }
        serve_media_file($f); // ONE serve helper, shared with action=media — see note there
    }

    // --- pending-sends: the outbound queue Pulse drains, with attachment filenames ---
    if ($action === 'pending-sends') {
        $s = $pdo->query(
            "SELECT m.id AS message_pk, m.platform, c.conversation_id, m.message_text, m.sender_name,
                    m.message_time, m.status
               FROM messages m JOIN conversations c ON c.id = m.conversation_pk
              WHERE m.direction='out' AND m.status='queued'
              ORDER BY m.id ASC");
        $rows = $s->fetchAll();
        if ($rows) {
            $ids = array_column($rows,'message_pk');
            $ph  = implode(',', array_fill(0,count($ids),'?'));
            $at  = $pdo->prepare("SELECT message_pk, cached_filename FROM attachments
                                   WHERE message_pk IN ($ph) AND cached_filename IS NOT NULL");
            $at->execute($ids);
            $by = [];
            foreach ($at->fetchAll() as $a) $by[$a['message_pk']][] = $a['cached_filename'];
            foreach ($rows as &$r) $r['attachments'] = $by[$r['message_pk']] ?? [];
            unset($r);
        }
        echo json_encode(['status'=>'ok','pending'=>$rows]); exit;
    }

    // --- send-result: Pulse reports what happened to one queued row ---
    // ⛔ TRANSITIONS (Solace, decided): queued|failed -> sent|failed is allowed, because a retry that
    // succeeds after a failure is a REAL path and must be recordable. 'sent' -> anything is REFUSED to
    // $warn, never applied: un-sending is not a real path. A result for an already-settled row is a
    // warning, not an error — it must not look like a delivery failure to the caller.
    if ($action === 'send-result') {
        $b = json_decode(file_get_contents('php://input'), true);
        if (!is_array($b)) { http_response_code(400); echo json_encode(['status'=>'error','error'=>'bad json']); exit; }
        $mpk = (int)($b['message_pk'] ?? 0);
        $st  = (string)($b['status'] ?? '');
        if ($mpk <= 0 || !in_array($st, ['sent','failed'], true)) {
            http_response_code(400); echo json_encode(['status'=>'error','error'=>"status must be 'sent' or 'failed', message_pk required"]); exit;
        }
        $w = [];
        $cur = $pdo->prepare("SELECT status, direction FROM messages WHERE id=?");
        $cur->execute([$mpk]); $row = $cur->fetch();
        if ($row === false) { http_response_code(404); echo json_encode(['status'=>'error','error'=>'no such message']); exit; }
        if ($row['direction'] !== 'out') { http_response_code(400); echo json_encode(['status'=>'error','error'=>'not an outbound message']); exit; }
        if ($row['status'] === 'sent') {
            $w[] = "message_pk $mpk is already 'sent'; result ignored (settled rows are never re-opened)";
            echo json_encode(['status'=>'ok','applied'=>false,'warnings'=>$w]); exit;
        }
        if (!in_array($row['status'], ['queued','failed'], true)) {
            $w[] = "message_pk $mpk is '{$row['status']}', not a sendable state; result ignored";
            echo json_encode(['status'=>'ok','applied'=>false,'warnings'=>$w]); exit;
        }
        $ext = trim((string)($b['external_id'] ?? ''));
        $u = $pdo->prepare("UPDATE messages SET status=?, external_id=COALESCE(NULLIF(?,''), external_id) WHERE id=?");
        $u->execute([$st, $ext, $mpk]);
        emit_sse('social_inbound', ['message_pk'=>$mpk,'status'=>$st,'direction'=>'out']);
        echo json_encode(['status'=>'ok','applied'=>true,'message_pk'=>$mpk,'new_status'=>$st,'warnings'=>$w]); exit;
    }
}

// --- STAFF lane: per-person Citadel token, NEVER the shared fleet key -------------------------
// These are actions a human takes from the app, so they authenticate as that human (Solace ruling
// (a), 2026-09-27). ⛔ The fleet key must not reach them: it is shared, server-side, and carries no
// identity, and every row written here records WHO wrote it.
if (in_array($action, $STAFF_ACTIONS, true)) {
    if (($_SERVER['REQUEST_METHOD'] ?? '') !== 'POST') {
        http_response_code(405); echo json_encode(['status'=>'error','error'=>'method']); exit;
    }
    $uid = citadel_user_or_401();   // returns the authkeys user_id, or exits 401

    if ($action === 'send') {
        $b = json_decode(file_get_contents('php://input'), true);
        if (!is_array($b)) { http_response_code(400); echo json_encode(['status'=>'error','error'=>'bad json']); exit; }
        $plat = normplat($b['platform'] ?? '');
        if (!$plat) { http_response_code(400); echo json_encode(['status'=>'error','error'=>'platform must be FB or IG']); exit; }
        $cid  = trim((string)($b['conversation_id'] ?? ''));
        if ($cid === '') { http_response_code(400); echo json_encode(['status'=>'error','error'=>'conversation_id required']); exit; }
        $text = (string)($b['message_text'] ?? '');
        $atts = is_array($b['attachments'] ?? null) ? array_values($b['attachments']) : [];
        // ⛔ Empty text is LEGAL if there is an attachment -- an attachment-only reply is a real thing,
        // exactly as it is inbound (all three of Ray's test frames are attachment-only). Empty AND no
        // attachment is not a message.
        if (trim($text) === '' && $atts === []) {
            http_response_code(400); echo json_encode(['status'=>'error','error'=>'message_text or at least one attachment required']); exit;
        }
        $warn = [];

        // Attachments here are filenames ALREADY uploaded via action=upload -- no fetching happens on
        // the send path. Validate the NAME shape and that the bytes actually exist, so a queued row can
        // never reference something Pulse will fail to fetch.
        $files = [];
        foreach ($atts as $fn) {
            $fn = basename((string)$fn);
            if (!preg_match('/^(avatar|att)_[0-9a-f]{16}$/', $fn)) {
                http_response_code(400); echo json_encode(['status'=>'error','error'=>"bad attachment filename: $fn"]); exit;
            }
            if (!is_file('/var/www/data.finestoflines.net/social_2026/media/' . $fn)) {
                http_response_code(400); echo json_encode(['status'=>'error','error'=>"attachment not present on disk: $fn"]); exit;
            }
            $files[] = $fn;
        }

        // --- ATTRIBUTION: artists.artist_nickname via the token's user_id (Solace's batch-2 ruling:
        // the shop's own voice -- "Nikki", not "Nicole Boggs"). NULL renders as nothing; NEVER invent.
        // ⚠️ PROVISIONAL GUARD, flagged to Solace 2026-09-29 and hers to replace: artists row 1 is a
        // SEED row whose fname, lname and nickname are all literally 'default c artist'. Attributing a
        // send to that string would be the forged-plausible-value class -- and it is worse than cosmetic
        // here, because Pulse's outgoing prefix format is "<id> <name> ~ text", so a placeholder would
        // be sent TO A CUSTOMER. Structural test: nickname == fname == lname identifies exactly that row
        // in current data (every real artist differs, including id 54 whose lname is blank).
        // `artists.status` is NOT usable for this -- the seed row carries status='1', same as Nikki.
        $sender_name = null;
        try {
            $DBL = $GLOBALS['DB_CONFIG']['production'];
            $al = new PDO("mysql:host={$DBL['host']};dbname={$DBL['lineup']};charset=utf8mb4", $DBL['user'], $DBL['pass'],
                [PDO::ATTR_ERRMODE=>PDO::ERRMODE_EXCEPTION, PDO::ATTR_DEFAULT_FETCH_MODE=>PDO::FETCH_ASSOC]);
            $st = $al->prepare("SELECT artist_nickname n, artist_fname f, artist_lname l FROM artists WHERE artist_id=?");
            $st->execute([$uid]);
            $ar = $st->fetch();
            if ($ar === false) {
                $warn[] = "no artists row for user_id $uid; sending unattributed (service accounts render as nothing)";
            } elseif ($ar['n'] === $ar['f'] && $ar['f'] === $ar['l']) {
                $warn[] = "artists row $uid looks like a seed placeholder ('{$ar['n']}'); sending unattributed rather than inventing a label";
            } elseif (trim((string)$ar['n']) === '') {
                $warn[] = "artists row $uid has an empty nickname; sending unattributed";
            } else {
                $sender_name = mb_substr(trim($ar['n']), 0, 191);
            }
        } catch (Throwable $e) {
            // ⛔ Attribution failing must NEVER block a staff reply. Unattributed is honest; refusing to
            // send because we could not look up a label would be the wrong failure by a wide margin.
            $warn[] = 'attribution lookup unavailable; sending unattributed';
        }

        $pdo->beginTransaction();
        try {
            $cq = $pdo->prepare("SELECT id, last_message_at FROM conversations WHERE platform=? AND conversation_id=? FOR UPDATE");
            $cq->execute([$plat, $cid]);
            $cr = $cq->fetch(PDO::FETCH_ASSOC);
            if ($cr === false) {
                $pdo->rollBack();
                http_response_code(404); echo json_encode(['status'=>'error','error'=>'no such conversation']); exit;
            }
            $conv_pk = (int)$cr['id'];
            $now_utc = gmdate('Y-m-d H:i:s');

            $digest = $files === [] ? '' : hash('sha256', count($files) . "\x1f" . implode("\x1e", (function($a){ sort($a, SORT_STRING); return $a; })($files)));
            $mi = $pdo->prepare("INSERT INTO messages
                (conversation_pk,platform,external_id,message_time,message_time_raw,body_hash,message_text,
                 direction,status,staff_user_id,sender_name,attachment_digest)
                VALUES (?,?,NULL,?,?,?,?, 'out','queued',?,?,?)");
            $mi->execute([$conv_pk,$plat,$now_utc,$now_utc,hash('sha256',$text),$text,(string)$uid,$sender_name,$digest]);
            $msg_pk = (int)$pdo->lastInsertId();

            foreach ($files as $fn) {
                $pdo->prepare("INSERT INTO attachments (message_pk,type,remote_url,cached_filename,fetch_error) VALUES (?,?,NULL,?,NULL)")
                    ->execute([$msg_pk, 'photo', $fn]);
            }

            // Rollup: an outbound message IS the newest activity, so the preview moves (SMS-list
            // convention). ⛔ unread_count is NEVER touched -- our own reply is not unread customer traffic.
            if ($now_utc > (string)$cr['last_message_at']) {
                $pdo->prepare("UPDATE conversations SET last_message_at=?, last_message_preview=? WHERE id=?")
                    ->execute([$now_utc, mb_substr($text !== '' ? $text : '[attachment]', 0, 158), $conv_pk]);
            }
            $pdo->commit();
        } catch (Throwable $e) {
            if ($pdo->inTransaction()) $pdo->rollBack();
            http_response_code(500); echo json_encode(['status'=>'error','error'=>'send failed']); exit;
        }

        // Existing event names only -- the frontend registers listeners BY NAME (e74518c).
        emit_sse('social_inbound', ['conversation_id'=>$cid,'platform'=>$plat,'direction'=>'out','status'=>'queued']);
        echo json_encode(['status'=>'ok','message_pk'=>$msg_pk,'status_value'=>'queued',
            'sender_name'=>$sender_name,'attachments'=>$files,'warnings'=>$warn]); exit;
    }

    // ⛔⛔ FAIL CLOSED. A name listed in $STAFF_ACTIONS with no handler above MUST NOT fall out of this
    // block: the POST auth gate below deliberately SKIPS staff actions, so falling through lands in the
    // INGEST path with a per-person token and no fleet key -- i.e. a staff token could forge inbound
    // customer messages. MEASURED, not theorised: with 'upload' declared-but-unimplemented, a POST of
    // action=upload carrying an ingest payload returned created_conversation:true. Closed 2026-09-29.
    // ⇒ Declaring an action and implementing it are two steps, and the gap between them was an auth hole.
    http_response_code(501);
    echo json_encode(['status'=>'error','error'=>"action '$action' is declared but not implemented"]); exit;
}

// The fleet WRITE key is required for POST (ingest). GET reads authenticate per-person below.
if (($_SERVER['REQUEST_METHOD'] ?? '') === 'POST' && $caller === null && !in_array($action, $STAFF_ACTIONS, true)) {
    http_response_code(401); echo json_encode(['status'=>'error','error'=>'auth']); exit;
}
if (($_SERVER['REQUEST_METHOD'] ?? '') === 'GET') {
    // READ auth (Solace ruling (a), 2026-09-27): the browser presents its OWN Citadel token,
    // validated per-person against the live authkeys table — NO shared key is ever compiled into
    // the frontend bundle. The fleet X-API-Key above is WRITE-only (Pulse, server-side).
    citadel_user_or_401();
    if ($action === 'conversations') {
        echo json_encode(['status'=>'ok','conversations'=>$pdo->query(
            "SELECT * FROM conversations ORDER BY last_message_at DESC")->fetchAll()]); exit;
    }
    if ($action === 'messages') {
        $cid = $_GET['conversation_id'] ?? ''; $plat = normplat($_GET['platform'] ?? '');
        $s = $pdo->prepare("SELECT m.* FROM messages m JOIN conversations c ON c.id=m.conversation_pk
            WHERE c.platform=? AND c.conversation_id=? ORDER BY m.message_time ASC");
        $s->execute([$plat,$cid]);
        $msgs = $s->fetchAll();
        if ($msgs) {
            $ids  = array_column($msgs,'id');
            $ph   = implode(',', array_fill(0,count($ids),'?'));
            $at   = $pdo->prepare("SELECT message_pk,type,cached_filename,fetch_error
                                   FROM attachments WHERE message_pk IN ($ph)");
            $at->execute($ids);
            $byMsg = [];
            foreach ($at->fetchAll() as $a) $byMsg[$a['message_pk']][] = [
                'type'=>$a['type'], 'file'=>$a['cached_filename'], 'error'=>$a['fetch_error'],
            ];
            foreach ($msgs as &$m) $m['attachments'] = $byMsg[$m['id']] ?? [];
            unset($m);
        }
        $cn = $pdo->prepare("SELECT contact_name FROM conversations WHERE platform=? AND conversation_id=?");
        $cn->execute([$plat,$cid]);
        echo json_encode(['status'=>'ok','messages'=>$msgs,'contact_name'=>($cn->fetchColumn() ?: null)]); exit;
    }

    if ($action === 'media') {
        $f = basename((string)($_GET['file'] ?? ''));               // basename: no traversal
        if (!preg_match('/^(avatar|att)_[0-9a-f]{16}$/', $f)) {     // exactly what fetch_media() mints
            http_response_code(400); echo json_encode(['status'=>'error','error'=>'bad file name']); exit;
        }
        serve_media_file($f);   // ONE serve helper, shared with outbound-media (see the function)
    }
    http_response_code(400); echo json_encode(['status'=>'error','error'=>'unknown action']); exit;
}

if (($_SERVER['REQUEST_METHOD'] ?? '') !== 'POST') { http_response_code(405); echo json_encode(['status'=>'error','error'=>'method']); exit; }

$p = json_decode(file_get_contents('php://input'), true);
if (!is_array($p)) { http_response_code(400); echo json_encode(['status'=>'error','error'=>'bad json']); exit; }
$platform = normplat($p['platform'] ?? '');
if (!$platform) { http_response_code(400); echo json_encode(['status'=>'error','error'=>'platform must be FB or IG']); exit; }
$conv_id = trim((string)($p['conversation_id'] ?? ''));
if ($conv_id === '') { http_response_code(400); echo json_encode(['status'=>'error','error'=>'conversation_id required']); exit; }

$mt_raw   = trim((string)($p['message_time'] ?? ''));
$mt_utc   = to_utc($mt_raw);
$det_raw  = trim((string)($p['detected_at'] ?? ''));
$body     = (string)($p['message_text'] ?? '');
$body_h   = hash('sha256', $body);
$ext_id   = ($x = trim((string)($p['message_id'] ?? ''))) !== '' ? $x : null;
$att_in   = is_array($p['attachments'] ?? null) ? $p['attachments'] : [];
$warn     = [];

// --- v2 DIRECTION. Validate against exactly {in,out}; anything else is a 400, NEVER a silent default.
// An unknown direction filed as 'in' would be a WRONG ROW THAT LOOKS RIGHT. The column is already
// enum('in','out'), so the engine would refuse too -- this is the friendly early error, not the only guard.
$direction = strtolower(trim((string)($p['direction'] ?? 'in')));
if ($direction === '') $direction = 'in';
if ($direction !== 'in' && $direction !== 'out') {
    http_response_code(400);
    echo json_encode(['status'=>'error','error'=>"direction must be 'in' or 'out'"]); exit;
}

// --- v2 BACKFILL. Historical import. Suppresses unread badges and SSE; never the row itself.
$backfill = filter_var($p['backfill'] ?? false, FILTER_VALIDATE_BOOLEAN);

// --- v2 SENDER_NAME. Pulse's payload VERBATIM -- what the Meta inbox actually SHOWED. Rare by design.
// ⛔ out-only, and a stray one on an inbound WARNS rather than 400s. The asymmetry is deliberate
// (Solace's rule): direction 400s because a bad value corrupts the ROW; sender_name is bad METADATA on a
// valid customer message, and rejecting it would discard a real customer's words over a label.
// REJECT WHAT CORRUPTS, WARN ON WHAT MERELY CONFUSES.
// ⛔ Never derive this from `artists` here: the ingest path has no Citadel token and no user_id to join
// on -- Pulse authenticates with the fleet key. That JOIN belongs to action=send.
$sender_name = null;
$sn_raw = trim((string)($p['sender_name'] ?? ''));
if ($sn_raw !== '') {
    if ($direction === 'out') $sender_name = mb_substr($sn_raw, 0, 191);
    else $warn[] = 'sender_name supplied with direction=in; ignored (attribution is out-only)';
}

// --- ATTACHMENT DIGEST (dedupe step 2). Pure computation from the payload: no DB, no network.
// Identity is the URL PATH BASENAME. Measured 2026-09-29 on the live rows:
//   NOT the full URL -- the CDN edge host varies within one conversation (scontent-dfw6-2 / -1 /
//     cdn.fbsbx.com) and oh/oe/_nc_ohc/ccb rotate on EVERY render. Verified end-to-end: same asset,
//     different host, all query params rotated -> IDENTICAL digest. That invariance is the requirement.
//   NOT type+count -- too coarse; that WAS the live defect (two different photos, same minute, empty
//     body => the second silently dropped as a duplicate).
// '' when there are no attachments (matches COALESCE(attachment_digest,'') for legacy NULL rows, so a
// message re-reported across the rollout boundary still collides); a hash otherwise, with count() inside
// it so "one attachment we cannot identify" stays distinct from "no attachments".
// Order-independence (A,B == B,A) is INTENDED.
$att_ident = [];
foreach ($att_in as $a) {
    $u    = is_array($a) ? ($a['url'] ?? null) : (is_string($a) ? $a : null);
    $path = (is_string($u) && $u !== '') ? parse_url($u, PHP_URL_PATH) : null;
    $att_ident[] = is_string($path) ? basename($path) : '';
}
sort($att_ident, SORT_STRING);
$att_digest = $att_ident === []
    ? ''
    : hash('sha256', count($att_ident) . "\x1f" . implode("\x1e", $att_ident));

// --- v2 HOIST, part 1: DEDUPE PRE-CHECK *BEFORE* ANY NETWORK FETCH.
// A re-run backfill must not re-download every photo it already has.
// ⛔ This is an OPTIMISATION, never the guarantee: the engine UNIQUE indexes still enforce inside the
// transaction below, so losing a race here costs one wasted fetch, never a duplicate row.
if ($ext_id !== null) {
    $pre = $pdo->prepare("SELECT conversation_pk FROM messages WHERE platform=? AND external_id=? LIMIT 1");
    $pre->execute([$platform, $ext_id]);
    $pre_pk = $pre->fetchColumn();
    if ($pre_pk !== false) {
        echo json_encode(['status'=>'ok','duplicate'=>true,'conversation_pk'=>(int)$pre_pk,
            'warnings'=>$warn,'prechecked'=>true]); exit;
    }
}

// --- v2 HOIST, part 2: EVERY NETWORK FETCH HAPPENS HERE, ABOVE THE TRANSACTION.
// BEFORE: fetch_media ran INSIDE the txn while holding SELECT ... FOR UPDATE on the conversation row.
// At CURLOPT_TIMEOUT=12 / CONNECTTIMEOUT=6 per call, one avatar + N attachments could hold that row lock
// for 12 + N*12 seconds ACROSS NETWORK I/O, blocking live inbound for the same conversation behind it.
// Backfill was the worst case by construction: its URLs are the most likely to be dead or slow.
// AFTER: fetch into locals here, then open the txn and do pure DB work with the lock held for ms.
$cx = $pdo->prepare("SELECT id FROM conversations WHERE platform=? AND conversation_id=? LIMIT 1");
$cx->execute([$platform, $conv_id]);
$conv_seen = ($cx->fetchColumn() !== false);
// Unlocked pre-check ONLY to decide whether an avatar fetch is worth doing. Worst case of losing this
// race is ONE unneeded avatar fetch -- cheap and idempotent. The LOCKED select below remains authoritative.
$av = $conv_seen ? ['file'=>null,'error'=>null] : fetch_media($p['avatar_url'] ?? null, 'avatar');

$att_fetched = [];
foreach ($att_in as $a) {
    $url = is_array($a) ? ($a['url'] ?? null) : (is_string($a) ? $a : null);
    $ty  = is_array($a) ? ($a['type'] ?? 'other') : 'other';
    $ty  = in_array($ty,['photo','video','file','audio']) ? $ty : 'other';
    $f   = fetch_media($url, 'att');
    $att_fetched[] = ['type'=>$ty,'url'=>$url,'file'=>$f['file'],'error'=>$f['error']];
}

// --- SORT: (platform, conversation_id) alone decides new-vs-existing. NEVER first_contact, NEVER contact_name. ---
$pdo->beginTransaction();
try {
    $s = $pdo->prepare("SELECT id, last_message_at FROM conversations WHERE platform=? AND conversation_id=? FOR UPDATE");
    $s->execute([$platform,$conv_id]);
    $crow      = $s->fetch(PDO::FETCH_ASSOC);
    $conv_pk   = ($crow === false) ? false : (int)$crow['id'];
    $prev_last = ($crow === false) ? null  : $crow['last_message_at'];
    $created   = false;

    // --- v2 GUARD 2: UNREAD. Outbound is not unread customer traffic, and a 30-day import must not
    // flood badges. ⚠️ This also replaces the conversation INSERT's HARDCODED 1 below -- same rule, and
    // that hardcoded literal is the half that is easy to miss.
    $unread_inc = ($direction === 'in' && !$backfill) ? 1 : 0;

    if ($conv_pk === false) {
        $ins = $pdo->prepare("INSERT INTO conversations
            (platform,conversation_id,account_ref,contact_name,contact_username,avatar_url,avatar_cached,last_message_at,last_message_preview,unread_count)
            VALUES (?,?,?,?,?,?,?,?,?,?)");
        $ins->execute([$platform,$conv_id,$p['account_ref']??null,$p['contact_name']??null,
            (($u=trim((string)($p['contact_username']??'')))!==''?$u:null),
            $p['avatar_url']??null,$av['file'], $mt_utc, mb_substr($body,0,158), $unread_inc]);
        $conv_pk = (int)$pdo->lastInsertId(); $created = true;
    } else {
        // first_contact is Pulse's belief; our store holds the fact. Use it only as an integrity WARNING.
        if (($p['first_contact'] ?? null) === true) $warn[] = 'first_contact=true but conversation already exists (upstream restart or reissued id)';
    }

    // --- DEDUPE at the engine (rule A): try insert, catch the unique violation ---
    $dup = false; $msg_pk = null;
    try {
        $mi = $pdo->prepare("INSERT INTO messages
            (conversation_pk,platform,external_id,message_time,message_time_raw,detected_at,body_hash,message_text,direction,sender_name,attachment_digest)
            VALUES (?,?,?,?,?,?,?,?,?,?,?)");
        $mi->execute([$conv_pk,$platform,$ext_id,$mt_utc,$mt_raw,($det_raw!==''?to_utc($det_raw):null),
            $body_h,$body,$direction,$sender_name,$att_digest]);
        $msg_pk = (int)$pdo->lastInsertId();
    } catch (PDOException $e) {
        if ($e->errorInfo[1] == 1062) { $dup = true; } else { throw $e; }
    }

    if ($dup) {
        // ⚠️ REGRESSION THE HOIST WOULD OTHERWISE INTRODUCE: we now fetch BEFORE knowing this is a
        // duplicate, so without this the bytes would be orphaned in media/ forever. fetch_media mints a
        // fresh random name per call, so these files are OURS ALONE and unlinking them cannot affect any
        // existing row. (Pre-hoist, the dup check ran before the fetch loop and the problem did not exist.)
        foreach ($att_fetched as $a) {
            if (!empty($a['file'])) @unlink('/var/www/data.finestoflines.net/social_2026/media/' . $a['file']);
        }
        $pdo->commit();
        echo json_encode(['status'=>'ok','duplicate'=>true,'conversation_pk'=>(int)$conv_pk,'warnings'=>$warn]); exit;
    }

    // attachments -- bytes were fetched above the transaction; this loop is pure DB.
    foreach ($att_fetched as $a) {
        $pdo->prepare("INSERT INTO attachments (message_pk,type,remote_url,cached_filename,fetch_error) VALUES (?,?,?,?,?)")
            ->execute([$msg_pk,$a['type'],$a['url'],$a['file'],$a['error']]);
    }

    // --- v2 GUARD 1: RECENCY. A 6-month-old backfilled message must NOT become the list preview; a
    // newest OUT message SHOULD (same convention as the SMS list). last_message_at is NOT NULL, so the
    // 'Y-m-d H:i:s' string comparison is safe. If the row was just created it already holds this
    // message's values, so only the not-created branch needs to move anything.
    if (!$created) {
        $is_newer = ($mt_utc !== null) && ($prev_last === null || $mt_utc > $prev_last);
        if ($is_newer) {
            $pdo->prepare("UPDATE conversations SET last_message_at=?, last_message_preview=?, unread_count=unread_count+? WHERE id=?")
                ->execute([$mt_utc, mb_substr($body,0,158), $unread_inc, $conv_pk]);
        } elseif ($unread_inc) {
            $pdo->prepare("UPDATE conversations SET unread_count=unread_count+? WHERE id=?")
                ->execute([$unread_inc, $conv_pk]);
        }
    }
    $pdo->commit();
} catch (Throwable $e) {
    if ($pdo->inTransaction()) $pdo->rollBack();
    http_response_code(500); echo json_encode(['status'=>'error','error'=>'ingest failed']); exit;
}

// --- v2 GUARD 3: SSE for LIVE traffic only. A 200-message import must not fire 200 refetches.
// ⛔ EXISTING EVENT NAMES for outbound too -- do NOT mint social_outbound. The frontend registers
// listeners BY NAME and an unknown name is silently dropped until someone adds a listener; that exact
// class cost us the live-update bug this morning (e74518c). The payload carries direction if we ever
// want it; the NAME stays in the registered set.
if (!$backfill) {
    emit_sse($created ? 'social_new_conversation' : 'social_inbound',
        ['conversation_id'=>$conv_id,'platform'=>$platform,'direction'=>$direction]);
}

echo json_encode(['status'=>'ok','duplicate'=>false,'created_conversation'=>$created,
    'conversation_pk'=>(int)$conv_pk,'message_pk'=>(int)$msg_pk,'warnings'=>$warn]);

// ---- helpers ----

/**
 * serve_media_file — THE ONLY PLACE BYTES LEAVE THIS ENDPOINT.
 * Shared deliberately by action=media (Citadel, per-person) and action=outbound-media (fleet key).
 * ⛔ DO NOT FORK THIS. The two callers authenticate differently and authorize differently, but the
 * SERVING rules — MIME allow-list, nosniff, Content-Length, private caching — must be identical.
 * Forking it is how one of the two paths quietly loses a header and nobody notices for months.
 * Callers are responsible for authorization BEFORE calling; this function only validates the NAME
 * and serves. It never decides who may read.
 */
function serve_media_file(string $f): void {
    if (!preg_match('/^(avatar|att)_[0-9a-f]{16}$/', $f)) {   // defence in depth: callers check too
        http_response_code(400); echo json_encode(['status'=>'error','error'=>'bad file name']); exit;
    }
    $path = '/var/www/data.finestoflines.net/social_2026/media/' . $f;
    if (!is_file($path)) { http_response_code(404); echo json_encode(['status'=>'error','error'=>'not cached']); exit; }
    $mime = (new finfo(FILEINFO_MIME_TYPE))->file($path) ?: 'application/octet-stream';
    // These bytes came from a third-party URL fetch_media() did not choose. If a remote ever served
    // HTML, finfo sniffs text/html and we would hand SAME-ORIGIN HTML back from the host that holds
    // Citadel tokens -> stored XSS. Sniffing still decides the type for real media; anything else is
    // forced inert.
    if (!preg_match('#^(image|video|audio)/#', $mime)) {
        $mime = 'application/octet-stream';
        header('Content-Disposition: attachment; filename="' . $f . '"');  // $f is regex-bound, no header injection
    }
    header('Content-Type: ' . $mime);
    header('X-Content-Type-Options: nosniff');                  // never let the browser re-sniff past us
    header('Content-Length: ' . filesize($path));
    header('Cache-Control: private, max-age=86400');            // token-gated => private, never shared
    readfile($path); exit;
}
function citadel_user_or_401(){
    // Accept "Authorization: Bearer <authkey>" or ?token=; validate against the live authkeys table.
    $h = $_SERVER['HTTP_AUTHORIZATION'] ?? '';
    $tok = (stripos($h,'bearer ')===0) ? trim(substr($h,7)) : trim((string)($_GET['token'] ?? ''));
    if ($tok==='') { http_response_code(401); echo json_encode(['status'=>'error','error'=>'missing token']); exit; }
    require_once '/var/www/data.finestoflines.net/config/db-config.php';
    $D=$GLOBALS['DB_CONFIG']['production'];
    try { $a=new PDO("mysql:host={$D['host']};dbname={$D['lineup']};charset=utf8mb4",$D['user'],$D['pass'],[PDO::ATTR_ERRMODE=>PDO::ERRMODE_EXCEPTION]);
        $st=$a->prepare('SELECT user_id FROM authkeys WHERE authkey=? AND expires > NOW()'); $st->execute([$tok]);
        $u=$st->fetchColumn();
    } catch(Throwable $e){ http_response_code(500); echo json_encode(['status'=>'error','error'=>'auth backend']); exit; }
    if ($u===false){ http_response_code(401); echo json_encode(['status'=>'error','error'=>'invalid or expired token']); exit; }
    return $u;
}
function normplat($v){ $v=strtolower(trim((string)$v));
    if($v==='fb'||$v==='facebook') return 'facebook';
    if($v==='ig'||$v==='instagram') return 'instagram'; return null; }
function to_utc($iso){ if($iso==='') return null;
    try{ $d=new DateTime($iso); $d->setTimezone(new DateTimeZone('UTC')); return $d->format('Y-m-d H:i:s'); }
    catch(Throwable $e){ return null; } }
// 2026-09-29: patch by Solace, landed by Zora, on Ray's order (Lineup-NX2). The old version
// @-suppressed mkdir + write and returned file=>name/error=>null UNCONDITIONALLY -> a silent
// cache failure recorded a forged success (fetch_error NULL) with no file on disk. Signed media
// URLs expire in ~32h (measured, not the ~30d we assumed), so a silent failure DESTROYS client
// media within a day. Every failure path now returns file=>null WITH a reason; temp+rename never
// publishes a partial write under the real name; is_writable catches the 09-27 condition pre-fetch.
function fetch_media($url,$kind){
    if(!$url) return ['file'=>null,'error'=>null];
    $dir='/var/www/data.finestoflines.net/social_2026/media';
    if(!is_dir($dir) && !@mkdir($dir,0755,true) && !is_dir($dir))
        return ['file'=>null,'error'=>'cache dir unavailable'];
    if(!is_writable($dir)) return ['file'=>null,'error'=>'cache dir not writable'];
    $ch=curl_init($url);
    curl_setopt_array($ch,[CURLOPT_RETURNTRANSFER=>true,CURLOPT_FOLLOWLOCATION=>true,CURLOPT_TIMEOUT=>12,CURLOPT_CONNECTTIMEOUT=>6]);
    $body=curl_exec($ch); $code=curl_getinfo($ch,CURLINFO_HTTP_CODE); curl_close($ch);
    if($body===false||$code>=400||$body==='') return ['file'=>null,'error'=>"fetch $code"];
    $name=$kind.'_'.bin2hex(random_bytes(8));
    $tmp="$dir/$name.part";
    $w=file_put_contents($tmp,$body);
    if($w===false || $w!==strlen($body)) { @unlink($tmp); return ['file'=>null,'error'=>'cache write failed']; }
    if(!rename($tmp,"$dir/$name"))       { @unlink($tmp); return ['file'=>null,'error'=>'cache rename failed']; }
    @chmod("$dir/$name",0644);
    return ['file'=>$name,'error'=>null];
}
function emit_sse($type,$data){
    $dir='/dev/shm/sse_lineup'; if(!is_dir($dir)) @mkdir($dir,0777,true);
    @file_put_contents("$dir/".uniqid('social_',true).'.json',
        json_encode(['device_id'=>'pulse-ingest','type'=>$type,'data'=>$data,'timestamp'=>time()]), LOCK_EX);
}