cayce-ops/runtime/editorials.mjs

// cayce-editorials — editoriales por burbuja (topic) + ingesta Nostr
//   1. ingesta     — kind 1 de relays por hashtag + cuentas seguidas → nostr_notes (🟢)
//   2. discovery   — pubkeys recurrentes → nostr_follows + kind 3 vía daemon (🟢 §6b)
//   3. editoriales — por topic activo: artículos 48h + notas Nostr → GLM → topic_editorials (🟢)
// "Lo no auditado no existió."

import pg from 'pg'
import { readFileSync } from 'node:fs'
import { loadIdentityBlock } from './outbox.mjs'
import { SimplePool } from 'nostr-tools/pool'
import { chat } from './llm.mjs'

const {
  ZAI_API_KEY,
  ZAI_BASE = 'https://api.z.ai/api/paas/v4/chat/completions',
  HARD_BACKEND_ENV = '/srv/hardcap-backend/.env',
  DAEMON = 'http://localhost:3011',
  PGURL,
} = process.env

if (!ZAI_API_KEY) { console.error('sin ZAI_API_KEY — falta EnvironmentFile en la unit'); process.exit(1) }

// DATABASE_URL del backend de hardcap (dueño de su DB)
const hardEnv = readFileSync(HARD_BACKEND_ENV, 'utf8')
const HC_URL = hardEnv.match(/^DATABASE_URL=(.+)$/m)?.[1]
if (!HC_URL) { console.error('sin DATABASE_URL en hardcap backend env'); process.exit(1) }
const hc = new pg.Pool({ connectionString: HC_URL })
const pgc = PGURL ? new pg.Pool({ connectionString: PGURL, max: 2 }) : null
const log = (zone, action, detail = {}, result = 'ok') =>
  pgc ? pgc.query('INSERT INTO actions_log (zone, action, detail, result) VALUES ($1,$2,$3,$4)', [zone, action, detail, result]) : Promise.resolve()
const pool = new SimplePool()

const RELAYS = ['wss://relay.damus.io', 'wss://nos.lol', 'wss://relay.primal.net']

// hashtags por categoría — fuente Nostr para cada burbuja
const HASHTAGS = {
  macro: ['bitcoin', 'grownostr', 'economy'],
  sovereignty: ['sovereignty', 'selfcustody', 'bitcoin'],
  institutional: ['bitcoin', 'etf', 'institutional'],
  protocol: ['bitcoin', 'lightning', 'nostr'],
  lightning: ['lightning', 'ln'],
  security: ['bitcoinsecurity', 'security', '5dollarwrench'],
  custody: ['selfcustody', 'multisig'],
  privacy: ['privacy', 'kychygiene'],
  mining: ['bitcoinmining', 'mining'],
  regulation: ['regulation', 'law'],
  latam: ['bitcoinlatam', 'argentina'],
  other: ['bitcoin'],
}
const VOICE = loadIdentityBlock('editorial')

async function glm(messages, maxTokens = 2000) {
  // cadena de fallback en llm.mjs; acá solo el pulido editorial + guards
  const raw = await chat(messages, maxTokens, { thinking: false })
  // reasoning filtrado: tomar el último párrafo no vacío y limpiar markdown de análisis
  const paras = raw.split('\n').map(s => s.trim()).filter(Boolean)
  let c = paras[paras.length - 1] ?? ''
  if (/^(\d+\.|[-*]|\*\*)/.test(c) && paras.length > 1) c = paras.filter(p => !/^(\d+\.|[-*]|\*\*Analyz)/.test(p)).join(' ')
  c = c.replace(/\*\*/g, '').trim()
  if (!c) throw new Error('glm devolvió contenido vacío')
  if (c.includes('async function') || c.includes('pgq(')) throw new Error('model returned code')
  return c
}

// ── 1. ingesta nostr: hashtags + seguidos, últimas 24h
async function ingestNostr() {
  const follows = (await hc.query('SELECT pubkey FROM nostr_follows WHERE active')).rows.map(r => r.pubkey)
  const since = Math.floor(Date.now() / 1000) - 86400
  // querySync de una query por vez: algunos relays rechazan arrays de filtros
  const queries = [{ kinds: [1], '#t': [...new Set(Object.values(HASHTAGS).flat())], since, limit: 200 }]
  if (follows.length) queries.push({ kinds: [1], authors: follows, since, limit: 200 })
  const evs = []
  for (const f of queries) {
    try { evs.push(...await pool.querySync(RELAYS, f, 12000)) } catch (e) { console.error('[ingesta] relay:', String(e).slice(0, 100)) }
  }
  let n = 0
  for (const ev of evs) {
    if (!ev.content || ev.content.length < 30) continue
    const tags = ev.tags.filter(t => t[0] === 't').map(t => t[1].toLowerCase())
    const hashtag = tags.find(t => Object.values(HASHTAGS).flat().includes(t)) || null
    try {
      const r = await hc.query(
        `INSERT INTO nostr_notes (event_id, pubkey, content, hashtag, published_at)
         VALUES ($1,$2,$3,$4,to_timestamp($5)) ON CONFLICT (event_id) DO NOTHING`,
        [ev.id, ev.pubkey, ev.content.slice(0, 2000), hashtag, ev.created_at])
      n += r.rowCount
    } catch { /* dup */ }
  }
  console.log(`[ingesta] ${n} notas nuevas (${evs.length} vistas, ${follows.length} seguidos)`)
  return n
}

// ── 2. discovery: pubkeys recurrentes se siguen (§6b, zona verde)
async function discoverFollows() {
  const { rows } = await hc.query(`
    SELECT pubkey, count(*) c FROM nostr_notes
    WHERE ingested_at > now() - interval '7 days'
    GROUP BY pubkey HAVING count(*) >= 3
      AND pubkey NOT IN (SELECT pubkey FROM nostr_follows)
    ORDER BY c DESC LIMIT 5`)
  const pk = '59ef3b5df26b92cb73986dc86af6294ca59a0a70f648ed0a92ccdb059fd42855'
  for (const r of rows) {
    if (r.pubkey === pk) continue
    await hc.query(`INSERT INTO nostr_follows (pubkey, category) VALUES ($1,'discovered') ON CONFLICT DO NOTHING`, [r.pubkey])
    console.log(`[discovery] siguiendo ${r.pubkey.slice(0, 8)} (${r.c} notas)`)
  }
  // kind 3 actualizado vía daemon
  const all = (await hc.query('SELECT pubkey FROM nostr_follows WHERE active')).rows.map(r => r.pubkey)
  if (all.length) {
    try {
      await fetch(`${DAEMON}/follows`, {
        method: 'POST', headers: { 'Content-Type': 'application/json' },
        body: JSON.stringify({ pubkeys: all }),
        signal: AbortSignal.timeout(30000),
      })
    } catch (e) { console.error('[follows]', String(e).slice(0, 120)) }
  }
}

// ── 3. editoriales por burbuja
async function generateEditorials() {
  // ventana 7d y umbral >=1: TODAS las burbujas con material tienen editorial (decisión supervisor)
  const { rows: bubbles } = await hc.query(`
    SELECT unnest(topics) AS topic, count(*) AS c,
           array_agg(title || ' — ' || coalesce(left(summary, 200), '')) AS items
    FROM articles
    WHERE status = 'PUBLISHED' AND "publishedAt" > now() - interval '7 days'
    GROUP BY 1 HAVING count(*) >= 1 ORDER BY c DESC`)
  let done = 0
  const failed = []
  for (const b of bubbles) {
    const tag = (HASHTAGS[b.topic] || ['bitcoin'])[0]
    const { rows: notes } = await hc.query(`
      SELECT left(content, 200) c FROM nostr_notes
      WHERE hashtag = $1 AND published_at > now() - interval '7 days'
      ORDER BY published_at DESC LIMIT 8`, [tag])
    const material = b.items.slice(0, 8).map(t => `- ${t}`).join('\n')
    const nostrPart = notes.length ? `\n\nNostr chatter (followed accounts + hashtags):\n${notes.map(n => `- ${n.c}`).join('\n')}` : ''
    try {
      const editorial = await glm([
        { role: 'user', content: `${VOICE}\n\nCategory: ${b.topic} (${b.c} articles in 48h)\n\nArticles:\n${material}${nostrPart}` },
      ])
      await hc.query(`
        INSERT INTO topic_editorials (topic, editorial, article_count, nostr_notes, generated_at)
        VALUES ($1,$2,$3,$4,now())
        ON CONFLICT (topic) DO UPDATE SET editorial=$2, article_count=$3, nostr_notes=$4, generated_at=now()`,
        [b.topic, editorial, b.c, notes.length])
      console.log(`[editorial] ${b.topic} (${b.c} art, ${notes.length} nostr) → ${editorial.slice(0, 60)}...`)
      done++
    } catch (e) {
      failed.push(b.topic)
      console.error(`[editorial] ${b.topic}:`, String(e).slice(0, 120))
    }
  }
  console.log(`[editoriales] ${done}/${bubbles.length} generadas`)
  await log('green', 'editorials_cycle', { done, total: bubbles.length, failed, error: failed.length ? 'Error: ' + (failed.length === bubbles.length ? 'todas las burbujas fallaron' : `${failed.length}/${bubbles.length} burbujas fallaron`) : undefined }, failed.length ? 'error' : 'ok')
  return { done, failed }
}

// ── health check de las webs (revisión periódica del agente — decisión supervisor 10-04)
async function healthCheck() {
  const sites = ['https://cayce-co.xyz', 'https://hardcap.news', 'https://bitcoinhotels.org'] // status queda fuera: basicauth 401 esperado
  for (const site of sites) {
    try {
      const r = await fetch(site, { signal: AbortSignal.timeout(15000) })
      if (r.status !== 200) {
        await log('red', 'site_health', { site, status: r.status, error: `site ${new URL(site).hostname} → HTTP ${r.status}`, at: new Date().toISOString() }, 'error')
        console.log(`[health] ${site} → ${r.status} 🔴`)
      } else {
        console.log(`[health] ${site} → 200 🟢`)
      }
    } catch (e) {
      await log('red', 'site_health', { site, error: `site ${new URL(site).hostname} → ${String(e).slice(0, 80)}`, at: new Date().toISOString() }, 'error')
      console.log(`[health] ${site} → ERROR 🔴`)
    }
  }
}

const cmd = process.argv[2] || 'all'
let outcome = { done: 0, failed: [] }
if (cmd === 'all') {
  await healthCheck()
  await ingestNostr()
  await discoverFollows()
  outcome = await generateEditorials()
} else if (cmd === 'editorials') outcome = await generateEditorials()
else if (cmd === 'ingest') { await ingestNostr(); await discoverFollows() }

// lo no auditado no existió — y lo que falla no sale en verde:
// total-failure = exit 1 (systemd en rojo, la retro lo ve)
if (pgc) await pgc.end()
await hc.end()
process.exit(outcome.failed.length === outcome.done + outcome.failed.length && outcome.failed.length > 0 ? 1 : 0)