@@ -0,0 +1,293 @@
/**
* arcrun km_wiki_ingest_drainer —— 無人值守卡片 ingest 編排 Worker（ ）
*
* 這是 workflow `km_wiki_ingest_drain` 的「具體 runtime」：workflow.yaml 是宣告式 spec，
* 本 Worker 是可真部署、有游標/觸發的執行體。復用已 live 的零件與服務，不動 cypher-executor：
* Gitea 抓卡 md → arcrun-code 零件（sandbox inline JS 解析）→ /entries 冪等 upsert（ ）
* → graph /triplets/ingest（
*
* 兩觸發模式共用同一 card-processing 核心（processCard）：
* (A) Phase 0 cron drain：
* 游標（存 KBDB 一個 cursor entry）之後取一小批 → 逐卡 processCard → 進游標（到底回捲）。
* 冪等：未改卡的 entry(content_hash)/triplet(uri+hash) 皆 skip，只有新/改卡真寫。
* (B) 穩態 webhook：
* commit 動到的 system-dev/wiki/cards/**.md）→ 逐卡 processCard。不重掃全庫。
*
* HTTP 路由（fetch）：
* GET / 健康
* GET /cursor 看游標狀態
* POST /drain 手動觸發一批 drain（
* POST /webhook Gitea push webhook 入口（delta 模式）
* POST /cursor/reset 游標歸零（重新全 drain 用）
*/
import { CARD_TO_ENVELOPE_USERCODE } from './usercode' ;
export interface Env {
REPO : string ; // Leo/notes
REF : string ; // main
OWNER : string ; // leo（ ）
CARDS_ROOT : string ; // system-dev/wiki/cards
CODE_WORKER_URL : string ; // https://arcrun-code.leo21c.workers.dev/
KBDB_URL : string ; // https://arcrun-kbdb.leo21c.workers.dev
GRAPH_URL : string ; // https://kbdb-graph-plugin.leo21c.workers.dev
GRAPH_KEY : string ; // leo
BATCH_SIZE : string ; // "2"
GITEA_TOKEN : string ; // secret：讀 repo
GITEA_WEBHOOK_SECRET? : string ; // secret（選填）：驗 webhook 簽章
}
const CURSOR_TYPE = 'ingest_cursor' ;
const cursorPageName = ( env : Env ) = > ` cursor:km_wiki_ingest_drain: ${ env . REPO } ` ;
// ── Gitea ────────────────────────────────────────────────────────────────────
const giteaApi = ( env : Env , path : string ) = > ` https://git.uncle6.me/api/v1/repos/ ${ env . REPO } ${ path } ` ;
/** 列出 CARDS_ROOT 下所有卡片（git tree recursive，排除 00-INDEX/.gitkeep），回 [{path, sha}] 已排序。 */
async function listCards ( env : Env ) : Promise < { path : string ; sha : string } [ ] > {
const r = await fetch ( giteaApi ( env , ` /git/trees/ ${ encodeURIComponent ( env . REF ) } ?recursive=true ` ) , {
headers : { Authorization : ` token ${ env . GITEA_TOKEN } ` } ,
} ) ;
if ( ! r . ok ) throw new Error ( ` gitea tree ${ r . status } : ${ ( await r . text ( ) ) . slice ( 0 , 200 ) } ` ) ;
const data = ( await r . json ( ) ) as { tree ? : { path : string ; type : string ; sha : string } [ ] ; truncated? : boolean } ;
const root = env . CARDS_ROOT . replace ( /\/$/ , '' ) + '/' ;
const cards = ( data . tree || [ ] )
. filter ( ( e ) = > e . type === 'blob' && e . path . startsWith ( root ) && e . path . endsWith ( '.md' ) )
. filter ( ( e ) = > {
const base = e . path . split ( '/' ) . pop ( ) || '' ;
return base !== '.gitkeep' && ! base . startsWith ( '00-INDEX' ) ;
} )
. map ( ( e ) = > ( { path : e.path , sha : e.sha } ) )
. sort ( ( a , b ) = > ( a . path < b . path ? - 1 : a.path > b . path ? 1 : 0 ) ) ;
// truncated：
( cards as unknown as { truncated? : boolean } ) . truncated = data . truncated === true ;
return cards ;
}
async function giteaRaw ( env : Env , relPath : string ) : Promise < string > {
const url = giteaApi ( env , ` /raw/ ${ relPath . split ( '/' ) . map ( encodeURIComponent ) . join ( '/' ) } ?ref= ${ encodeURIComponent ( env . REF ) } ` ) ;
const r = await fetch ( url , { headers : { Authorization : ` token ${ env . GITEA_TOKEN } ` } } ) ;
if ( ! r . ok ) throw new Error ( ` gitea raw ${ relPath } ${ r . status } ` ) ;
return r . text ( ) ;
}
// ── card-processing 核心 ──────────────────────────────────────────────────────
async function parseCard ( env : Env , md : string , relPath : string ) : Promise < any > {
const r = await fetch ( env . CODE_WORKER_URL , {
method : 'POST' ,
headers : { 'content-type' : 'application/json' } ,
body : JSON.stringify ( {
code : CARD_TO_ENVELOPE_USERCODE ,
input : { md , relPath , repo : env.REPO , opts : { budget : 40 } } ,
limits : { timeout_ms : 3000 , max_output_bytes : 4 * 1024 * 1024 } ,
} ) ,
} ) ;
const out = ( await r . json ( ) ) as { success : boolean ; data? : any ; error? : string } ;
if ( ! out . success ) throw new Error ( ` code 零件解析失敗 ${ relPath } : ${ out . error } ` ) ;
return out . data ; // { entry, envelopes, meta, nodeCount, tripletCount }
}
function metaHash ( entry : any ) : string | undefined {
try {
return typeof entry . metadata_json === 'string' ? JSON . parse ( entry . metadata_json ) . content_hash : undefined ;
} catch {
return undefined ;
}
}
async function upsertEntry ( env : Env , entry : any ) : Promise < { action : string ; id? : string } > {
const pageName = entry . page_name ;
const metadata_json = JSON . stringify ( entry . metadata ) ;
const newHash = entry . metadata . content_hash ;
const lookup = await fetch (
` ${ env . KBDB_URL } /entries?page_name= ${ encodeURIComponent ( pageName ) } &owner_id= ${ encodeURIComponent ( env . OWNER ) } ` ,
) ;
const existing = ( ( await lookup . json ( ) ) as { entries? : any [ ] } ) . entries ? . [ 0 ] ;
if ( existing ) {
if ( metaHash ( existing ) === newHash ) return { action : 'skip' , id : existing.id } ;
const r = await fetch ( ` ${ env . KBDB_URL } /entries/ ${ existing . id } ` , {
method : 'PATCH' ,
headers : { 'content-type' : 'application/json' } ,
body : JSON.stringify ( { content : entry.content , tags_json : JSON.stringify ( entry . tags || [ ] ) , metadata_json } ) ,
} ) ;
return { action : 'patch' , id : existing.id , . . . ( r . ok ? { } : { action : 'patch-failed' } ) } ;
}
const r = await fetch ( ` ${ env . KBDB_URL } /entries ` , {
method : 'POST' ,
headers : { 'content-type' : 'application/json' } ,
body : JSON.stringify ( {
entry_type : entry.entry_type ,
content : entry.content ,
page_name : pageName ,
owner_id : env.OWNER ,
tags_json : JSON.stringify ( entry . tags || [ ] ) ,
source : entry.metadata.source ,
metadata_json ,
} ) ,
} ) ;
const out = ( await r . json ( ) ) as { entry ? : { id? : string } } ;
return { action : 'create' , id : out.entry?.id } ;
}
async function postEnvelopes ( env : Env , envelopes : any [ ] ) : Promise < any [ ] > {
const results = [ ] ;
for ( const envp of envelopes ) {
// envelopes 已由 code 節點剝除 _-鍵；此處再保險剝一次。
const clean = Object . fromEntries ( Object . entries ( envp ) . filter ( ( [ k ] ) = > ! k . startsWith ( '_' ) ) ) ;
const r = await fetch ( ` ${ env . GRAPH_URL } /triplets/ingest?owner_id= ${ encodeURIComponent ( env . OWNER ) } ` , {
method : 'POST' ,
headers : { 'content-type' : 'application/json' , 'X-Arcrun-API-Key' : env . GRAPH_KEY } ,
body : JSON.stringify ( clean ) ,
} ) ;
const text = await r . text ( ) ;
let body : any ;
try { body = JSON . parse ( text ) ; } catch { body = text ; }
results . push ( { status : r.status , uri : ( envp as any ) . source ? . uri , triplets : ( envp as any ) . triplets ? . length , body } ) ;
}
return results ;
}
/** 一張卡的完整處理（冪等）：fetch → parse → upsert entry → post envelopes。 */
async function processCard ( env : Env , relPath : string ) : Promise < any > {
const md = await giteaRaw ( env , relPath ) ;
const plan = await parseCard ( env , md , relPath ) ;
const entry = await upsertEntry ( env , plan . entry ) ;
const envelopes = await postEnvelopes ( env , plan . envelopes ) ;
const wrote = entry . action !== 'skip' || envelopes . some ( ( e ) = > e . body && e . body . skipped === false ) ;
return {
relPath ,
page_name : plan.entry.page_name ,
content_hash : ( plan . entry . metadata . content_hash || '' ) . slice ( 0 , 12 ) ,
entry : entry.action ,
envelopes : envelopes.map ( ( e ) = > ( { status : e.status , triplets : e.triplets , skipped : e.body?.skipped , ingested : e.body?.ingested } ) ) ,
wrote ,
} ;
}
// ── 游標（存 KBDB 一個 ingest_cursor entry；不 embed）─────────────────────────
interface CursorState { last_path : string ; cycle : number ; processed_total : number ; updated_at : number ; }
async function getCursor ( env : Env ) : Promise < { id? : string ; state : CursorState } > {
const r = await fetch (
` ${ env . KBDB_URL } /entries?page_name= ${ encodeURIComponent ( cursorPageName ( env ) ) } &owner_id= ${ encodeURIComponent ( env . OWNER ) } ` ,
) ;
const e = ( ( await r . json ( ) ) as { entries? : any [ ] } ) . entries ? . [ 0 ] ;
if ( e ) {
try { return { id : e.id , state : JSON.parse ( e . content ) } ; } catch { /* fallthrough */ }
return { id : e.id , state : { last_path : '' , cycle : 0 , processed_total : 0 , updated_at : 0 } } ;
}
return { state : { last_path : '' , cycle : 0 , processed_total : 0 , updated_at : 0 } } ;
}
async function setCursor ( env : Env , cur : { id? : string ; state : CursorState } ) : Promise < void > {
const content = JSON . stringify ( cur . state ) ;
if ( cur . id ) {
await fetch ( ` ${ env . KBDB_URL } /entries/ ${ cur . id } ` , {
method : 'PATCH' , headers : { 'content-type' : 'application/json' } , body : JSON.stringify ( { content } ) ,
} ) ;
} else {
await fetch ( ` ${ env . KBDB_URL } /entries ` , {
method : 'POST' , headers : { 'content-type' : 'application/json' } ,
body : JSON.stringify ( { entry_type : CURSOR_TYPE , page_name : cursorPageName ( env ) , owner_id : env.OWNER , content ,
metadata_json : JSON.stringify ( { kind : 'ingest_cursor' , embed : false } ) } ) ,
} ) ;
}
}
/** cron / 手動觸發：drain 一小批。游標之後取 BATCH 張；到底回捲（continuous drain）。 */
async function drainBatch ( env : Env , batchSize : number ) : Promise < any > {
const cards = await listCards ( env ) ;
const truncated = ( cards as unknown as { truncated? : boolean } ) . truncated ;
const cur = await getCursor ( env ) ;
let idx = cards . findIndex ( ( c ) = > c . path > cur . state . last_path ) ;
let wrapped = false ;
if ( idx < 0 ) { idx = 0 ; wrapped = true ; } // 到底 → 回捲重掃（靠冪等，未改卡 cheap skip）
const batch = cards . slice ( idx , idx + batchSize ) ;
const results = [ ] ;
for ( const c of batch ) results . push ( await processCard ( env , c . path ) ) ;
const newLast = batch . length ? batch [ batch . length - 1 ] . path : cur.state.last_path ;
cur . state = {
last_path : newLast ,
cycle : cur.state.cycle + ( wrapped ? 1 : 0 ) ,
processed_total : cur.state.processed_total + batch . length ,
updated_at : Math.floor ( Date . now ( ) / 1000 ) ,
} ;
await setCursor ( env , cur ) ;
return {
mode : 'cron-drain' , total_cards : cards.length , truncated : ! ! truncated , wrapped ,
batch : batch.map ( ( c ) = > c . path ) , results ,
wrote : results.filter ( ( r ) = > r . wrote ) . length , skipped : results.filter ( ( r ) = > ! r . wrote ) . length ,
cursor : cur.state ,
} ;
}
// ── webhook（
async function verifyGiteaSig ( env : Env , raw : string , sig : string | null ) : Promise < boolean > {
if ( ! env . GITEA_WEBHOOK_SECRET ) return true ; // 未設 secret → 不驗（leo 自有 repo）
if ( ! sig ) return false ;
const key = await crypto . subtle . importKey ( 'raw' , new TextEncoder ( ) . encode ( env . GITEA_WEBHOOK_SECRET ) ,
{ name : 'HMAC' , hash : 'SHA-256' } , false , [ 'sign' ] ) ;
const mac = await crypto . subtle . sign ( 'HMAC' , key , new TextEncoder ( ) . encode ( raw ) ) ;
const hex = [ . . . new Uint8Array ( mac ) ] . map ( ( b ) = > b . toString ( 16 ) . padStart ( 2 , '0' ) ) . join ( '' ) ;
return hex === sig ;
}
async function handleWebhook ( env : Env , raw : string ) : Promise < any > {
const payload = JSON . parse ( raw ) as { commits ? : { added? : string [ ] ; modified? : string [ ] ; removed? : string [ ] } [ ] } ;
const root = env . CARDS_ROOT . replace ( /\/$/ , '' ) + '/' ;
const isCard = ( p : string ) = > p . startsWith ( root ) && p . endsWith ( '.md' ) && ! ( p . split ( '/' ) . pop ( ) || '' ) . startsWith ( '00-INDEX' ) ;
const changed = new Set < string > ( ) ;
const removed = new Set < string > ( ) ;
for ( const cm of payload . commits || [ ] ) {
for ( const p of [ . . . ( cm . added || [ ] ) , . . . ( cm . modified || [ ] ) ] ) if ( isCard ( p ) ) changed . add ( p ) ;
for ( const p of cm . removed || [ ] ) if ( isCard ( p ) ) removed . add ( p ) ;
}
const results = [ ] ;
for ( const p of changed ) results . push ( await processCard ( env , p ) ) ;
return {
mode : 'webhook-delta' , changed : [ . . . changed ] , removed : [ . . . removed ] ,
note : removed.size ? 'removed 卡的清理未實作（graph 有 deprecate 機制，另議）' : undefined ,
results , wrote : results.filter ( ( r ) = > r . wrote ) . length ,
} ;
}
// ── Worker 入口 ────────────────────────────────────────────────────────────────
export default {
async scheduled ( _event : ScheduledEvent , env : Env , ctx : ExecutionContext ) : Promise < void > {
ctx . waitUntil ( drainBatch ( env , Number ( env . BATCH_SIZE || '2' ) ) . then ( ( ) = > { } ) . catch ( ( ) = > { } ) ) ;
} ,
async fetch ( req : Request , env : Env , _ctx : ExecutionContext ) : Promise < Response > {
const url = new URL ( req . url ) ;
const json = ( o : unknown , s = 200 ) = > new Response ( JSON . stringify ( o , null , 2 ) , { status : s , headers : { 'content-type' : 'application/json' } } ) ;
try {
if ( req . method === 'GET' && url . pathname === '/' ) {
return json ( { ok : true , service : 'km_wiki_ingest_drainer' , repo : env.REPO , cards_root : env.CARDS_ROOT , batch : env.BATCH_SIZE } ) ;
}
if ( req . method === 'GET' && url . pathname === '/cursor' ) {
return json ( await getCursor ( env ) ) ;
}
if ( req . method === 'POST' && url . pathname === '/cursor/reset' ) {
const cur = await getCursor ( env ) ;
cur . state = { last_path : '' , cycle : 0 , processed_total : 0 , updated_at : Math.floor ( Date . now ( ) / 1000 ) } ;
await setCursor ( env , cur ) ;
return json ( { ok : true , cursor : cur.state } ) ;
}
if ( req . method === 'POST' && url . pathname === '/drain' ) {
const n = Number ( url . searchParams . get ( 'batch' ) || env . BATCH_SIZE || '2' ) ;
return json ( await drainBatch ( env , n ) ) ;
}
if ( req . method === 'POST' && url . pathname === '/webhook' ) {
const raw = await req . text ( ) ;
const ok = await verifyGiteaSig ( env , raw , req . headers . get ( 'X-Gitea-Signature' ) ) ;
if ( ! ok ) return json ( { error : 'bad signature' } , 401 ) ;
// Gitea ping event（測試）→ 回 pong
if ( req . headers . get ( 'X-Gitea-Event' ) === 'ping' ) return json ( { ok : true , pong : true } ) ;
return json ( await handleWebhook ( env , raw ) ) ;
}
return json ( { error : 'not found' } , 404 ) ;
} catch ( e ) {
return json ( { error : e instanceof Error ? e.message : String ( e ) } , 500 ) ;
}
} ,
} ;
kbdb-cc