Rebuild: schema for groups, rule sets, ranges, release catalog, chunk store, fragments, probe queue, rule bookkeeping and nodes

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

database/migrations/1770000000100_create_groups.ts

@@ -0,0 +1,27 @@
+import { BaseSchema } from '@adonisjs/lucid/schema'
+
+/**
+ * The indexed groups (REBUILD.md, group list). `priority` decides a
+ * crosspost's owner: the indexed group in its Xref with the lowest number.
+ * New groups get numbers after the existing ones, so they never take
+ * articles from groups already done.
+ */
+export default class extends BaseSchema {
+  async up() {
+    this.schema.createTable('groups', (table) => {
+      table.increments('id')
+      table.string('name', 255).notNullable().unique()
+      table.integer('priority').notNullable().unique()
+      table.boolean('active').notNullable().defaultTo(true)
+      // From GROUP when planned: the server's oldest and newest article.
+      table.bigInteger('low').nullable()
+      table.bigInteger('high').nullable()
+      table.timestamp('planned_at', { useTz: true }).nullable()
+      table.timestamp('created_at', { useTz: true }).notNullable().defaultTo(this.now())
+    })
+  }
+
+  async down() {
+    this.schema.dropTable('groups')
+  }
+}

database/migrations/1770000000200_create_rule_sets.ts

@@ -0,0 +1,21 @@
+import { BaseSchema } from '@adonisjs/lucid/schema'
+
+/**
+ * Versions of what workers drop (REBUILD.md, rules and recovery): dropped
+ * signatures and tools, the per-article limit, the sample share. Every range
+ * records the version it ran with.
+ */
+export default class extends BaseSchema {
+  async up() {
+    this.schema.createTable('rule_sets', (table) => {
+      table.integer('version').primary()
+      table.jsonb('rules').notNullable()
+      table.text('note').nullable()
+      table.timestamp('created_at', { useTz: true }).notNullable().defaultTo(this.now())
+    })
+  }
+
+  async down() {
+    this.schema.dropTable('rule_sets')
+  }
+}

database/migrations/1770000000300_create_ranges.ts

@@ -0,0 +1,38 @@
+import { BaseSchema } from '@adonisjs/lucid/schema'
+
+/**
+ * A group's history in leasable pieces (REBUILD.md, coordinator):
+ * planned → (spooled by the prefetcher) → leased → done. `next_id` is the
+ * resume point: everything before it has been stored.
+ */
+export default class extends BaseSchema {
+  async up() {
+    this.schema.createTable('ranges', (table) => {
+      table.bigIncrements('id')
+      table.integer('group_id').notNullable().references('id').inTable('groups').onDelete('CASCADE')
+      table.bigInteger('start_id').notNullable()
+      table.bigInteger('end_id').notNullable()
+      table.bigInteger('next_id').notNullable()
+      table.string('status', 8).notNullable().defaultTo('planned')
+      // Lower first; the coordinator plans the oldest era first.
+      table.integer('priority').notNullable().defaultTo(0)
+      table.integer('rule_version').nullable()
+      table.integer('node_id').nullable()
+      table.integer('attempts').notNullable().defaultTo(0)
+      // Chunks of this range in the spool (prefetch); all of them when spooled_at is set.
+      table.integer('spooled_chunks').notNullable().defaultTo(0)
+      table.timestamp('spooled_at', { useTz: true }).nullable()
+      table.timestamp('leased_at', { useTz: true }).nullable()
+      table.timestamp('heartbeat_at', { useTz: true }).nullable()
+      table.timestamp('done_at', { useTz: true }).nullable()
+      table.bigInteger('headers').notNullable().defaultTo(0)
+      table.text('error').nullable()
+      table.unique(['group_id', 'start_id'])
+    })
+    this.schema.raw(`CREATE INDEX ranges_todo_idx ON ranges (priority, id) WHERE status IN ('planned', 'leased')`)
+  }
+
+  async down() {
+    this.schema.dropTable('ranges')
+  }
+}

database/migrations/1770000000400_create_releases.ts

@@ -0,0 +1,41 @@
+import { BaseSchema } from '@adonisjs/lucid/schema'
+
+/**
+ * The release catalog (REBUILD.md, part 4): one row per release, the hot
+ * part on fast disk. Updated with aggregates once per batch; the files
+ * themselves are in release_chunks.
+ */
+export default class extends BaseSchema {
+  async up() {
+    this.schema.createTable('releases', (table) => {
+      table.bigIncrements('id')
+      table.string('release_key', 40).notNullable().unique()
+      table.text('name').notNullable()
+      table.string('source', 16).notNullable()
+      table.text('poster').nullable()
+      table.jsonb('groups').notNullable()
+      table.timestamp('posted_at', { useTz: true }).nullable()
+      // Files announced by the [i/N] counter, when present.
+      table.integer('file_total').nullable()
+      table.integer('files').notNullable().defaultTo(0)
+      table.bigInteger('segments').notNullable().defaultTo(0)
+      table.bigInteger('bytes').notNullable().defaultTo(0)
+      table.integer('chunks').notNullable().defaultTo(0)
+      table.bigInteger('first_article').nullable()
+      table.bigInteger('last_article').nullable()
+      table.boolean('readable').notNullable().defaultTo(false)
+      table.boolean('has_par2').notNullable().defaultTo(false)
+      table.boolean('complete').nullable()
+      table.timestamp('settled_at', { useTz: true }).nullable()
+      table.timestamp('first_seen_at', { useTz: true }).notNullable().defaultTo(this.now())
+      table.timestamp('last_seen_at', { useTz: true }).notNullable().defaultTo(this.now())
+    })
+    this.schema.raw(`CREATE EXTENSION IF NOT EXISTS pg_trgm`)
+    this.schema.raw(`CREATE INDEX releases_name_trgm ON releases USING gin (name gin_trgm_ops)`)
+    this.schema.raw(`CREATE INDEX releases_posted_at_idx ON releases (posted_at DESC)`)
+  }
+
+  async down() {
+    this.schema.dropTable('releases')
+  }
+}

database/migrations/1770000000500_create_release_chunks.ts

@@ -0,0 +1,34 @@
+import { BaseSchema } from '@adonisjs/lucid/schema'
+import env from '#start/env'
+
+/**
+ * The chunk store (REBUILD.md, part 3; decision a): one write-once row per
+ * release per batch, its files encoded by lib/chunks. Partitioned by month
+ * of writing, in CHUNK_TABLESPACE (the HDD) when set; ChunkPartitions keeps
+ * the next months' partitions ready. A re-run of a range replaces its chunks
+ * (by range_id).
+ */
+export default class extends BaseSchema {
+  async up() {
+    const tablespace = env.get('CHUNK_TABLESPACE')
+    const where = tablespace ? ` TABLESPACE ${tablespace}` : ''
+    this.schema.raw(`
+      CREATE TABLE release_chunks (
+        id bigserial,
+        release_id bigint NOT NULL,
+        range_id bigint NOT NULL,
+        batch_seq integer NOT NULL,
+        files integer NOT NULL,
+        segments integer NOT NULL,
+        payload bytea NOT NULL,
+        created_at timestamptz NOT NULL DEFAULT now(),
+        PRIMARY KEY (id, created_at)
+      ) PARTITION BY RANGE (created_at)${where}`)
+    this.schema.raw(`CREATE INDEX release_chunks_release_idx ON release_chunks (release_id)`)
+    this.schema.raw(`CREATE INDEX release_chunks_range_idx ON release_chunks (range_id)`)
+  }
+
+  async down() {
+    this.schema.raw(`DROP TABLE IF EXISTS release_chunks CASCADE`)
+  }
+}

database/migrations/1770000000600_create_fragments.ts

@@ -0,0 +1,24 @@
+import { BaseSchema } from '@adonisjs/lucid/schema'
+
+/**
+ * Segments of files cut off by a range's edge (REBUILD.md, fragments): kept
+ * by range, merged when the neighbouring range is done, then written as a
+ * chunk.
+ */
+export default class extends BaseSchema {
+  async up() {
+    this.schema.createTable('fragments', (table) => {
+      table.bigIncrements('id')
+      table.bigInteger('range_id').notNullable()
+      table.string('file_key', 40).notNullable()
+      table.bigInteger('article_number').notNullable()
+      table.jsonb('segment').notNullable()
+      table.index(['file_key'])
+      table.index(['range_id'])
+    })
+  }
+
+  async down() {
+    this.schema.dropTable('fragments')
+  }
+}

database/migrations/1770000000700_create_probe_candidates.ts

@@ -0,0 +1,25 @@
+import { BaseSchema } from '@adonisjs/lucid/schema'
+
+/**
+ * The probe queue (REBUILD.md, part 6): per release the few articles phase 2
+ * should read. Phase 1 only fills it.
+ */
+export default class extends BaseSchema {
+  async up() {
+    this.schema.createTable('probe_candidates', (table) => {
+      table.bigIncrements('id')
+      table.bigInteger('release_id').notNullable()
+      table.string('kind', 8).notNullable()
+      table.text('message_id').notNullable()
+      table.text('file').notNullable()
+      table.string('status', 8).notNullable().defaultTo('waiting')
+      table.timestamp('created_at', { useTz: true }).notNullable().defaultTo(this.now())
+      table.unique(['release_id', 'kind', 'message_id'])
+    })
+    this.schema.raw(`CREATE INDEX probe_candidates_waiting_idx ON probe_candidates (id) WHERE status = 'waiting'`)
+  }
+
+  async down() {
+    this.schema.dropTable('probe_candidates')
+  }
+}

database/migrations/1770000000800_create_rule_bookkeeping.ts

@@ -0,0 +1,35 @@
+import { BaseSchema } from '@adonisjs/lucid/schema'
+
+/**
+ * What the rules dropped, per range (REBUILD.md, rules and recovery), and
+ * the sample of it kept in full.
+ */
+export default class extends BaseSchema {
+  async up() {
+    this.schema.createTable('rule_counts', (table) => {
+      table.bigInteger('range_id').notNullable()
+      table.string('rule', 12).notNullable()
+      table.text('key').notNullable()
+      table.bigInteger('articles').notNullable().defaultTo(0)
+      table.bigInteger('files').notNullable().defaultTo(0)
+      table.primary(['range_id', 'rule', 'key'])
+    })
+    this.schema.createTable('rule_samples', (table) => {
+      table.bigIncrements('id')
+      table.bigInteger('range_id').notNullable()
+      table.string('rule', 12).notNullable()
+      table.text('rule_key').notNullable()
+      table.string('release_key', 40).notNullable()
+      table.text('name').notNullable()
+      table.integer('files').notNullable()
+      table.integer('segments').notNullable()
+      table.binary('payload').notNullable()
+      table.index(['rule', 'rule_key'])
+    })
+  }
+
+  async down() {
+    this.schema.dropTable('rule_samples')
+    this.schema.dropTable('rule_counts')
+  }
+}

database/migrations/1770000000900_create_nodes.ts

@@ -0,0 +1,30 @@
+import { BaseSchema } from '@adonisjs/lucid/schema'
+
+/** Edge workers and their API tokens (the `node` guard). */
+export default class extends BaseSchema {
+  async up() {
+    this.schema.createTable('nodes', (table) => {
+      table.increments('id')
+      table.string('name', 64).notNullable().unique()
+      table.timestamp('last_seen_at', { useTz: true }).nullable()
+      table.timestamp('created_at', { useTz: true }).notNullable().defaultTo(this.now())
+    })
+    this.schema.createTable('node_access_tokens', (table) => {
+      table.increments('id')
+      table.integer('tokenable_id').notNullable().unsigned().references('id').inTable('nodes').onDelete('CASCADE')
+      table.string('type').notNullable()
+      table.string('name').nullable()
+      table.string('hash').notNullable()
+      table.text('abilities').notNullable()
+      table.timestamp('created_at')
+      table.timestamp('updated_at')
+      table.timestamp('last_used_at').nullable()
+      table.timestamp('expires_at').nullable()
+    })
+  }
+
+  async down() {
+    this.schema.dropTable('node_access_tokens')
+    this.schema.dropTable('nodes')
+  }
+}