Something went wrong. Try again.
Japanese learning that pulls JA Bluesky posts, scores them per-token, and serves them thricely.
Something went wrong. Try again.
4.8 kB · 173 lines
TypeScript
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174// Lite CRDB helpers for direct-ingest mode.// Mirrors packages/db/src/queries/ingest.ts using @easier-idx/core StoreOps.
import type { StoreOps } from "@easier-idx/core"
// ---------------------------------------------------------------------------// Ingest helpers// ---------------------------------------------------------------------------
export interface ReadingInsert { source: "bluesky" | "distinbox" source_ref: string source_uri: string | null source_cid: string | null author_did: string | null lang: string text: string created_at: string | null sentence_count: number token_count: number approx_tiktoken_count: number}
/** Insert a reading. Returns the new id, or null if ON CONFLICT (dedup). */export async function insertReading( ops: StoreOps, row: ReadingInsert): Promise<string | null> { const rows = await ops.query<{ id: string }>( `INSERT INTO readings (source, source_ref, source_uri, source_cid, author_did, lang, text, created_at, sentence_count, token_count, approx_tiktoken_count) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11) ON CONFLICT (source, source_ref) DO NOTHING RETURNING id`, [ row.source, row.source_ref, row.source_uri, row.source_cid, row.author_did, row.lang, row.text, row.created_at, row.sentence_count, row.token_count, row.approx_tiktoken_count, ] ) return rows[0]?.id ?? null}
export interface SentenceInsert { idx: number text: string begin: number end: number}
/** Bulk-insert sentences for a reading. Returns ordered id[]. */export async function insertSentences( ops: StoreOps, readingId: string, sentences: readonly SentenceInsert[]): Promise<string[]> { if (sentences.length === 0) return []
const cols = 5 const placeholders: string[] = [] const params: unknown[] = []
for (let i = 0; i < sentences.length; i++) { const off = i * cols placeholders.push( `($${off + 1}, $${off + 2}, $${off + 3}, $${off + 4}, $${off + 5})` ) const s = sentences[i]! params.push(readingId, s.idx, s.text, s.begin, s.end) }
const rows = await ops.query<{ id: string }>( `INSERT INTO sentences (reading_id, idx, text, begin, "end") VALUES ${placeholders.join(", ")} ON CONFLICT (reading_id, idx) DO NOTHING RETURNING id`, params ) return rows.map((r) => r.id)}
export interface SentenceTokenInsert { sentence_id: string idx: number surface: string reading_kata: string | null dictionary_form: string | null pos: string pos_full: string | null begin: number end: number}
// Postgres/CockroachDB cap a single statement at 65535 wire-protocol// parameters. With 9 columns per row that is ⌊65535/9⌋ = 7281 rows max;// 7000 leaves headroom for any future column additions. Long Japanese// posts (20K chars → ~10K morphemes) would otherwise overflow.const MAX_TOKENS_PER_INSERT = 7000
/** Bulk-insert sentence tokens. */export async function insertSentenceTokens( ops: StoreOps, tokens: readonly SentenceTokenInsert[]): Promise<void> { if (tokens.length === 0) return
for (let start = 0; start < tokens.length; start += MAX_TOKENS_PER_INSERT) { const chunk = tokens.slice(start, start + MAX_TOKENS_PER_INSERT) const cols = 9 const placeholders: string[] = [] const params: unknown[] = []
for (let i = 0; i < chunk.length; i++) { const off = i * cols placeholders.push( `($${off + 1}, $${off + 2}, $${off + 3}, $${off + 4}, $${off + 5}, $${off + 6}, $${off + 7}, $${off + 8}, $${off + 9})` ) const t = chunk[i]! params.push( t.sentence_id, t.idx, t.surface, t.reading_kata, t.dictionary_form, t.pos, t.pos_full, t.begin, t.end ) }
await ops.run( `INSERT INTO sentence_tokens (sentence_id, idx, surface, reading_kata, dictionary_form, pos, pos_full, begin, "end") VALUES ${placeholders.join(", ")} ON CONFLICT (sentence_id, idx) DO NOTHING`, params ) }}
/** Enqueue readings for embedding. Skips duplicates. */export async function enqueueEmbeddings( ops: StoreOps, rows: readonly { readingId: string; inputTokens: number }[]): Promise<void> { if (rows.length === 0) return const cols = 2 const placeholders: string[] = [] const params: unknown[] = []
for (let i = 0; i < rows.length; i++) { const off = i * cols placeholders.push(`($${off + 1}, $${off + 2})`) params.push(rows[i]!.readingId, rows[i]!.inputTokens) }
await ops.run( `INSERT INTO embedding_queue (reading_id, input_tokens) VALUES ${placeholders.join(", ")} ON CONFLICT (reading_id) DO NOTHING`, params )}