<?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;
// The fleet WRITE key is required for POST (ingest). GET reads authenticate per-person below.
if (($_SERVER['REQUEST_METHOD'] ?? '') === 'POST' && $caller === null) {
    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;
        }
        $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';
        // ZORA AMENDMENT 2026-09-29 (Solace's block otherwise verbatim, reported to her):
        // 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
        // (her intent, and it is what closes (c) on the serving path); anything else is forced inert.
        // All four cached files today are image/jpeg, so this changes no current behaviour.
        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;
    }
    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'] : [];

// --- SORT: (platform, conversation_id) alone decides new-vs-existing. NEVER first_contact, NEVER contact_name. ---
$pdo->beginTransaction();
try {
    $s = $pdo->prepare("SELECT id FROM conversations WHERE platform=? AND conversation_id=? FOR UPDATE");
    $s->execute([$platform,$conv_id]); $conv_pk = $s->fetchColumn();
    $created = false; $warn = [];
    if ($conv_pk === false) {
        $av = fetch_media($p['avatar_url'] ?? null, 'avatar');
        $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 (?,?,?,?,?,?,?,?,?,1)");
        $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)]);
        $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)';
    }

    // --- ATTACHMENT DIGEST (step 2 of the dedupe fix; Ray's go 2026-09-29 11:0x) ---
    // Folded into the dedup_key generated expression by Solace in step 3. Identity is the URL PATH
    // BASENAME, measured on the live rows 2026-09-29:
    //   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:
    //                         same asset, different host, all query params rotated -> IDENTICAL digest.
    //                         That invariance is the entire requirement.
    //   NOT type+count     -- too coarse; that WAS the live defect (two different photos, same minute,
    //                         empty body => the second was silently dropped as a duplicate).
    // ⚠️ '' WHEN THERE ARE NO ATTACHMENTS, a hash otherwise. Both halves are load-bearing:
    //   '' matches what COALESCE(attachment_digest,'') yields for legacy NULL rows, so the same message
    //      re-reported across the rollout boundary still collides (it would otherwise fire exactly once,
    //      during the transition, and never reproduce).
    //   a hash for >=1 keeps "one attachment we cannot identify" distinct from "no attachments" -- with
    //      count() omitted those two collapse to the same value, which is the drop-a-real-message
    //      direction again.
    // Order-independence (A,B == B,A) is INTENDED; do not "fix" it.
    $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));

    // --- 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,attachment_digest)
            VALUES (?,?,?,?,?,?,?,?, 'in', ?)");
        $mi->execute([$conv_pk,$platform,$ext_id,$mt_utc,$mt_raw,($det_raw!==''?to_utc($det_raw):null),$body_h,$body,$att_digest]);
        $msg_pk = (int)$pdo->lastInsertId();
    } catch (PDOException $e) {
        if ($e->errorInfo[1] == 1062) { $dup = true; } else { throw $e; }
    }

    if ($dup) {
        $pdo->commit();
        echo json_encode(['status'=>'ok','duplicate'=>true,'conversation_pk'=>(int)$conv_pk,'warnings'=>$warn]); exit;
    }

    // attachments (fetch-at-ingest; degrades to URL-only on fetch failure — see the Daystrom-egress gate in the row)
    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');
        $pdo->prepare("INSERT INTO attachments (message_pk,type,remote_url,cached_filename,fetch_error) VALUES (?,?,?,?,?)")
            ->execute([$msg_pk,$ty,$url,$f['file'],$f['error']]);
    }
    // conversation rollup
    $pdo->prepare("UPDATE conversations SET last_message_at=?, last_message_preview=?, unread_count=unread_count+1 WHERE id=?")
        ->execute([$mt_utc, mb_substr($body,0,158), $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;
}

// --- NOTIFY the Lineup UI, same mechanism as SMS (Solace's contract) ---
emit_sse($created ? 'social_new_conversation' : 'social_inbound',
    ['conversation_id'=>$conv_id,'platform'=>$platform]);

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

// ---- helpers ----
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);
}