--- id: 008-p2p-schema-and-config-127966 title: Add P2P tables, config flags and db helpers created: 2026-09-29 depends_on: [] est_files: 5 --- # 008 — P2P tables, config flags and db helpers ## Objective Lay the storage foundation for peer-to-peer sharing described in `docs/p2p-architecture.md` (READ IT FIRST — it is short). After this plan: - `server/p2p-config.js` exports `P2P` (config) and `loadP2pConfig(env)`. Defaults: **P2P enabled**, **malware scan disabled**, stale after 7 days. - `server/p2p-db.js` creates `p2p_content`, `p2p_devices`, `p2p_holders`, `video_views` and adds `media_cache.sha256`, with query helpers. - The server calls `initP2pSchema()` at boot. Nothing else changes behaviour yet. ## Context the executor must NOT rediscover - `server/db.js` exports `db` (libsql client) and `initDb()`. Its `MEDIA_COLS` set (~line 308) whitelists columns `upsertMedia()` may write: ```js const MEDIA_COLS = new Set([ 'status', 'gen', 'size', 'height', 'vcodec', 'acodec', 'duration', 'optimized', 'meta', 'priority', 'auto', 'attempts', 'error', 'retry_at', 'created_at', 'updated_at', 'last_access', 'hits', ]); ``` - `server/server.js` `main()` (~line 1916): `await initDb();` then `await media.init();`. - Server test conventions: `bun:test`, a temp `DB_PATH` set BEFORE importing `./db.js` (see `server/notes.test.js:1-12`), and each file runs in its own `bun test` process via the `server/package.json` "test" script (db.js is a singleton). ## Steps 1. Create `server/p2p-config.js` with exactly: ```js /* p2p-config.js — peer-to-peer settings (see docs/p2p-architecture.md). * P2P is ON unless P2P_ENABLED=0. The malware scan is OFF unless * P2P_MALWARE_SCAN=1. Hashing + validateMedia are never optional. */ import { dirname, join } from 'node:path'; const num = (v, d) => (Number.isFinite(Number(v)) && String(v).trim() !== '' ? Number(v) : d); export function loadP2pConfig(env = process.env) { const dbDir = dirname(env.DB_PATH || './data/ytplayer.db'); return { enabled: env.P2P_ENABLED !== '0', malwareScan: env.P2P_MALWARE_SCAN === '1', scanCmd: (env.P2P_SCAN_CMD || 'clamscan --no-summary --infected').trim(), staleDays: num(env.P2P_STALE_DAYS, 7), keepMinViews: num(env.P2P_KEEP_MIN_VIEWS, 3), keepDays: num(env.P2P_KEEP_DAYS, 30), keepRecentDays: num(env.P2P_KEEP_RECENT_DAYS, 14), intakeDir: env.P2P_INTAKE_DIR || join(dbDir, 'p2p-intake'), intakeMaxBytes: num(env.P2P_INTAKE_MAX_BYTES, 3 * 1024 ** 3), }; } export const P2P = loadP2pConfig(); ``` 2. Create `server/p2p-db.js` — copy VERBATIM from the "p2p-db.js" appendix at the end of this plan. 3. Create `server/p2p-db.test.js` — copy VERBATIM from the "p2p-db.test.js" appendix. 4. `server/db.js` — add `'sha256'` to the end of `MEDIA_COLS` (after `'hits'`). 5. `server/package.json` "test" script — append ` && bun test ./p2p-db.test.js`. 6. `server/server.js` — add `import { initP2pSchema } from './p2p-db.js';` next to the other local imports, and in `main()` directly after `await initDb();` add `await initP2pSchema();`. 7. `docker-compose.yml` — in the `ytplayer` service `environment:` block, after the `LYRICS_WORKER_TOKEN` line, add: ```yaml # Peer-to-peer sharing (docs/p2p-architecture.md). ON by default. P2P_ENABLED: "${P2P_ENABLED:-1}" # Malware scan before a file's hash is admitted. OFF by default; needs an # image built with INSTALL_CLAMAV=1. Hashing + media validation always run. P2P_MALWARE_SCAN: "${P2P_MALWARE_SCAN:-0}" # P2P_STALE_DAYS: "7" # holder shown as stale after this many days unchecked # P2P_KEEP_MIN_VIEWS: "3" # server keeps copies with ≥ this many views… # P2P_KEEP_DAYS: "30" # …in this many days # P2P_KEEP_RECENT_DAYS: "14" # …or played this recently ``` ## Out of scope / do NOT touch - No routes, no media-cache changes, no frontend changes (later plans). - Do not edit `initDb()`'s SQL; the new column is added by `initP2pSchema()`. ## Verification ```bash cd /home/user/ytplayer/server && bun test ./p2p-db.test.js 2>&1 | tail -4 bun run test 2>&1 | grep -E "^ *[0-9]+ (pass|fail)" bun build server.js --target=bun --outdir=/tmp/ytp-check >/dev/null && echo SERVER_OK ``` Expected: `4 pass 0 fail`; every file `0 fail`; `SERVER_OK`. ## Report format (executor: follow exactly) Output ONLY the following, no other prose: 1. `git diff` (unified) of all changes (new files in full). 2. Raw output of the Verification commands. 3. `Findings:` — max 10 lines. Do not commit. Do not push. Do not touch files outside the Steps. --- ## Appendix — p2p-db.js ```js /* ============================================================================ * p2p-db.js — tables + queries for peer-to-peer sharing * (docs/p2p-architecture.md). Shares the libsql client from db.js. * * p2p_content one row per verified file (cid = sha256 of the bytes); never * deleted, only revoked — the catalog grows over time * p2p_devices registered devices (secret stored as sha256) * p2p_holders which device holds which cid; PERSISTENT (no TTL) with * last_verified_at — the UI decides what is "stale" * video_views per-video per-day view counts (retention criteria) * All timestamps are ms epochs. * ========================================================================== */ import { db } from './db.js'; export async function initP2pSchema() { await db.executeMultiple(` CREATE TABLE IF NOT EXISTS p2p_content ( cid TEXT PRIMARY KEY, video_id TEXT NOT NULL, size INTEGER NOT NULL, height INTEGER NOT NULL DEFAULT 0, vcodec TEXT, acodec TEXT, duration REAL NOT NULL DEFAULT 0, meta TEXT NOT NULL DEFAULT '{}', origin TEXT NOT NULL, -- server | intake status TEXT NOT NULL DEFAULT 'verified', -- verified | revoked scan TEXT NOT NULL DEFAULT 'skipped', -- skipped | clean created_at INTEGER NOT NULL, verified_at INTEGER NOT NULL ); CREATE INDEX IF NOT EXISTS idx_p2p_content_video ON p2p_content (video_id, created_at DESC); CREATE TABLE IF NOT EXISTS p2p_devices ( device_id TEXT PRIMARY KEY, secret_hash TEXT NOT NULL, fingerprint TEXT, profile TEXT, share INTEGER NOT NULL DEFAULT 1, created_at INTEGER NOT NULL, last_seen_at INTEGER NOT NULL ); CREATE TABLE IF NOT EXISTS p2p_holders ( cid TEXT NOT NULL, device_id TEXT NOT NULL, status TEXT NOT NULL DEFAULT 'active', -- active | removed trust TEXT NOT NULL DEFAULT 'reported', -- reported | challenged first_reported_at INTEGER NOT NULL, last_verified_at INTEGER NOT NULL, removed_at INTEGER, PRIMARY KEY (cid, device_id) ); CREATE INDEX IF NOT EXISTS idx_p2p_holders_device ON p2p_holders (device_id, status); CREATE TABLE IF NOT EXISTS video_views ( video_id TEXT NOT NULL, day TEXT NOT NULL, n INTEGER NOT NULL DEFAULT 0, PRIMARY KEY (video_id, day) ); `); // media_cache.sha256 — the cid of the current ..mp4 (plan 009). try { await db.execute('ALTER TABLE media_cache ADD COLUMN sha256 TEXT'); } catch (e) { if (!/duplicate column/i.test(String(e.message))) throw e; } } const rowsOf = (r) => r.rows.map((row) => { const o = {}; r.columns.forEach((c, i) => { o[c] = row[i]; }); return o; }); // ---- content ---------------------------------------------------------------- export async function upsertContent(c) { await db.execute({ sql: `INSERT INTO p2p_content (cid, video_id, size, height, vcodec, acodec, duration, meta, origin, status, scan, created_at, verified_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, 'verified', ?, ?, ?) ON CONFLICT(cid) DO UPDATE SET verified_at = excluded.verified_at, scan = excluded.scan`, args: [c.cid, c.videoId, c.size, c.height || 0, c.vcodec || null, c.acodec || null, c.duration || 0, JSON.stringify(c.meta || {}), c.origin, c.scan || 'skipped', c.now, c.now], }); } export async function getContent(cid) { const r = await db.execute({ sql: 'SELECT * FROM p2p_content WHERE cid = ?', args: [cid] }); return rowsOf(r)[0] || null; } export async function listContentForVideo(videoId) { const r = await db.execute({ sql: "SELECT * FROM p2p_content WHERE video_id = ? AND status = 'verified' ORDER BY created_at DESC LIMIT 10", args: [videoId], }); return rowsOf(r); } export async function knownCids(cids) { if (!cids.length) return new Set(); const out = new Set(); for (let i = 0; i < cids.length; i += 200) { const part = cids.slice(i, i + 200); const r = await db.execute({ sql: `SELECT cid FROM p2p_content WHERE status = 'verified' AND cid IN (${part.map(() => '?').join(',')})`, args: part, }); for (const row of r.rows) out.add(row[0]); } return out; } export async function revokeContent(cid) { await db.execute({ sql: "UPDATE p2p_content SET status = 'revoked' WHERE cid = ?", args: [cid] }); } // ---- devices ------------------------------------------------------------------ export async function createDevice({ deviceId, secretHash, fingerprint, profile, now }) { await db.execute({ sql: `INSERT INTO p2p_devices (device_id, secret_hash, fingerprint, profile, share, created_at, last_seen_at) VALUES (?, ?, ?, ?, 1, ?, ?)`, args: [deviceId, secretHash, fingerprint || null, profile || null, now, now], }); } export async function getDevice(deviceId) { const r = await db.execute({ sql: 'SELECT * FROM p2p_devices WHERE device_id = ?', args: [deviceId] }); return rowsOf(r)[0] || null; } export async function touchDevice(deviceId, { now, share, profile } = {}) { await db.execute({ sql: `UPDATE p2p_devices SET last_seen_at = ?, share = COALESCE(?, share), profile = COALESCE(?, profile) WHERE device_id = ?`, args: [now, share === undefined ? null : (share ? 1 : 0), profile || null, deviceId], }); } // ---- holders (persistent; never expired by time) ------------------------------ export async function upsertHolder({ cid, deviceId, trust = 'reported', now }) { await db.execute({ sql: `INSERT INTO p2p_holders (cid, device_id, status, trust, first_reported_at, last_verified_at) VALUES (?, ?, 'active', ?, ?, ?) ON CONFLICT(cid, device_id) DO UPDATE SET status = 'active', removed_at = NULL, last_verified_at = excluded.last_verified_at, trust = CASE WHEN p2p_holders.trust = 'challenged' OR excluded.trust = 'challenged' THEN 'challenged' ELSE 'reported' END`, args: [cid, deviceId, trust, now, now], }); } export async function setHolderTrust({ cid, deviceId, trust, now }) { await db.execute({ sql: 'UPDATE p2p_holders SET trust = ?, last_verified_at = ? WHERE cid = ? AND device_id = ?', args: [trust, now, cid, deviceId], }); } export async function removeHolder({ cid, deviceId, now }) { await db.execute({ sql: "UPDATE p2p_holders SET status = 'removed', removed_at = ? WHERE cid = ? AND device_id = ? AND status = 'active'", args: [now, cid, deviceId], }); } // A full report: every active holding of this device NOT in `keep` is removed. export async function removeHoldersExcept({ deviceId, keep, now }) { const r = await db.execute({ sql: "SELECT cid FROM p2p_holders WHERE device_id = ? AND status = 'active'", args: [deviceId], }); const keepSet = new Set(keep); let removed = 0; for (const row of r.rows) { if (keepSet.has(row[0])) continue; await removeHolder({ cid: row[0], deviceId, now }); removed++; } return removed; } export async function activeHoldingsOf(deviceId) { const r = await db.execute({ sql: "SELECT cid FROM p2p_holders WHERE device_id = ? AND status = 'active'", args: [deviceId], }); return r.rows.map((row) => row[0]); } // Holders of one cid, joined with the device's share flag. Newest check first. export async function listHolders(cid, limit = 50) { const r = await db.execute({ sql: `SELECT h.device_id, h.trust, h.first_reported_at, h.last_verified_at, d.share FROM p2p_holders h JOIN p2p_devices d ON d.device_id = h.device_id WHERE h.cid = ? AND h.status = 'active' ORDER BY h.last_verified_at DESC LIMIT ?`, args: [cid, limit], }); return rowsOf(r); } // ---- views + stats -------------------------------------------------------------- export function dayKey(ms) { return new Date(ms).toISOString().slice(0, 10); } export async function addView(videoId, now) { await db.execute({ sql: `INSERT INTO video_views (video_id, day, n) VALUES (?, ?, 1) ON CONFLICT(video_id, day) DO UPDATE SET n = n + 1`, args: [videoId, dayKey(now)], }); } export async function viewsSince(videoId, sinceMs) { const r = await db.execute({ sql: 'SELECT COALESCE(SUM(n), 0) FROM video_views WHERE video_id = ? AND day >= ?', args: [videoId, dayKey(sinceMs)], }); return Number(r.rows[0][0]) || 0; } export async function p2pStats() { const one = async (sql) => Number((await db.execute(sql)).rows[0][0]) || 0; return { content: await one("SELECT COUNT(*) FROM p2p_content WHERE status = 'verified'"), revoked: await one("SELECT COUNT(*) FROM p2p_content WHERE status = 'revoked'"), devices: await one('SELECT COUNT(*) FROM p2p_devices'), holders: await one("SELECT COUNT(*) FROM p2p_holders WHERE status = 'active'"), heldCids: await one("SELECT COUNT(DISTINCT cid) FROM p2p_holders WHERE status = 'active'"), }; } ``` ## Appendix — p2p-db.test.js ```js // P2P tables against a real temp libsql DB (docs/p2p-architecture.md). import { test, expect, beforeAll } from 'bun:test'; import { mkdtempSync } from 'node:fs'; import { tmpdir } from 'node:os'; import { join } from 'node:path'; const root = mkdtempSync(join(tmpdir(), 'ytp-p2p-test-')); process.env.DB_PATH = join(root, 'test.db'); const dbmod = await import('./db.js'); const P = await import('./p2p-db.js'); const { loadP2pConfig } = await import('./p2p-config.js'); const CID = 'a'.repeat(64); const CID2 = 'b'.repeat(64); const T0 = Date.UTC(2026, 8, 29, 12); beforeAll(async () => { await dbmod.initDb(); await P.initP2pSchema(); await P.initP2pSchema(); // idempotent (ALTER TABLE duplicate column is ignored) }); test('config defaults: P2P on, malware scan off', () => { const c = loadP2pConfig({}); expect(c.enabled).toBe(true); expect(c.malwareScan).toBe(false); expect(c.staleDays).toBe(7); expect(loadP2pConfig({ P2P_ENABLED: '0', P2P_MALWARE_SCAN: '1' })).toMatchObject({ enabled: false, malwareScan: true }); }); test('content upsert/get/known/revoke', async () => { await P.upsertContent({ cid: CID, videoId: 'dQw4w9WgXcQ', size: 1000, height: 720, vcodec: 'h264', acodec: 'aac', duration: 212, meta: { title: 'x' }, origin: 'server', now: T0 }); const c = await P.getContent(CID); expect(c.video_id).toBe('dQw4w9WgXcQ'); expect(c.status).toBe('verified'); expect([...(await P.knownCids([CID, CID2]))]).toEqual([CID]); expect((await P.listContentForVideo('dQw4w9WgXcQ')).length).toBe(1); await P.upsertContent({ cid: CID2, videoId: 'dQw4w9WgXcQ', size: 5, origin: 'intake', now: T0 }); await P.revokeContent(CID2); expect([...(await P.knownCids([CID2]))]).toEqual([]); }); test('holders persist, never expire, and a full report removes missing ones', async () => { await P.createDevice({ deviceId: 'dev_1', secretHash: 'h', fingerprint: 'fp', now: T0 }); await P.upsertHolder({ cid: CID, deviceId: 'dev_1', now: T0 }); // 90 days later with no new report: still listed (UI marks it stale). let hs = await P.listHolders(CID); expect(hs.length).toBe(1); expect(hs[0].last_verified_at).toBe(T0); await P.setHolderTrust({ cid: CID, deviceId: 'dev_1', trust: 'challenged', now: T0 + 1000 }); await P.upsertHolder({ cid: CID, deviceId: 'dev_1', trust: 'reported', now: T0 + 2000 }); hs = await P.listHolders(CID); expect(hs[0].trust).toBe('challenged'); // a later plain report never downgrades trust expect(hs[0].last_verified_at).toBe(T0 + 2000); expect(await P.removeHoldersExcept({ deviceId: 'dev_1', keep: [], now: T0 + 3000 })).toBe(1); expect((await P.listHolders(CID)).length).toBe(0); expect(await P.activeHoldingsOf('dev_1')).toEqual([]); await P.upsertHolder({ cid: CID, deviceId: 'dev_1', now: T0 + 4000 }); // re-added expect(await P.activeHoldingsOf('dev_1')).toEqual([CID]); }); test('views per day and window sums', async () => { await P.addView('vid00000001', T0); await P.addView('vid00000001', T0); await P.addView('vid00000001', T0 - 40 * 86400_000); expect(await P.viewsSince('vid00000001', T0 - 30 * 86400_000)).toBe(2); expect(await P.viewsSince('vid00000001', T0 - 50 * 86400_000)).toBe(3); const s = await P.p2pStats(); expect(s.content).toBe(1); expect(s.devices).toBe(1); }); ``` ## Execution log - Executor: in-session Agent (haiku). Attempts: 1. Fix rounds: 0. - Orchestrator re-ran Verification: `p2p-db.js`, `p2p-db.test.js`, `p2p-config.js` byte-identical to the plan; p2p-db tests 4 pass; all 8 server test files 0 fail; `SERVER_OK`; server boots and `/api/version` returns 200. - Executor Findings (verbatim): All steps executed successfully. P2P schema, config, and tests created verbatim from plan appendices. Server builds without errors. No deviations from plan requirements.