An upload spread over groups is one release

FTD posts one upload's files to different groups ("Knowsum-Loops_Are_My_
Religion-WEB-2026-FTD": par2 and part1 to a.b.music.misc, part2 to
a.b.sounds.music, each article in one group only; releases 81825221 and
81825412): the catalog matched a chunk only to releases of its own
group. A chunk without one there now joins a release of its key in
another group whose window is within 3 hours, when none of its file
names are in that release yet; a post of the same files to another group
(other message-ids, same names) stays a release of its own, or its NZB
would hold every file twice. Functional tests for both.

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

app/services/batch_store.ts

@@ -2,6 +2,7 @@ import db from '@adonisjs/lucid/services/db'
 import type { TransactionClientContract } from '@adonisjs/lucid/types/database'
 import { UPLOAD_GAP_MS, type EdgeBatch, type ProbeCandidate, type ReleaseChunk } from '#lib/edge/index'
 import ChunkPartitions from '#services/chunk_partitions'
+import { decodeChunk } from '#lib/chunks/index'
 
 export class StaleBatch extends Error {}
 
@@ -101,6 +102,54 @@ class BatchStore {
     })
   }
 
+  /**
+   * An upload spread over groups (FTD: par2 and part1 to a.b.music.misc,
+   * part2 to a.b.sounds.music, each article in one group only; LOG.md
+   * 2026-10-11 (9)): a chunk without a release in its own group joins one of
+   * its key in another group whose window is within UPLOAD_GAP_MS, when none
+   * of its file names are in that release yet. A post of the same files to
+   * another group (other message-ids, the same names) stays a release of its
+   * own, or the NZB would hold every file twice. Reads the candidate's file
+   * names from its chunks; candidates are few.
+   */
+  private async joinAcrossGroups(
+    trx: TransactionClientContract,
+    part: ReleaseChunk[],
+    windows: { idx: number; key: string; first: number | null | undefined; last: number | null | undefined }[],
+    groupId: number,
+    joined: Map<number, number>
+  ) {
+    const open = windows.filter((w) => !joined.has(w.idx) && w.first)
+    if (open.length === 0) return
+    const { rows: candidates } = await trx.rawQuery(
+      `SELECT v.idx, r.id
+       FROM jsonb_to_recordset(?::jsonb) AS v(idx int, key text, first bigint, last bigint)
+       JOIN releases r ON r.identity_key = v.key AND r.group_id <> ?
+       WHERE r.posted_at <= to_timestamp((v.last + ?) / 1000.0) AND coalesce(r.last_posted_at, r.posted_at) >= to_timestamp((v.first - ?) / 1000.0)
+       ORDER BY v.idx, r.last_posted_at DESC NULLS LAST, r.id DESC`,
+      [JSON.stringify(open), groupId, UPLOAD_GAP_MS, UPLOAD_GAP_MS]
+    )
+    if (candidates.length === 0) return
+    const ids: number[] = [...new Set<number>(candidates.map((c: { id: string }) => Number(c.id)))]
+    const names = new Map<number, Set<string>>()
+    const { rows: chunks } = await trx.rawQuery(`SELECT release_id, payload FROM release_chunks WHERE release_id = ANY(?)`, [ids])
+    for (const c of chunks) {
+      const set = names.get(Number(c.release_id)) ?? new Set<string>()
+      for (const f of decodeChunk(c.payload)) set.add(f.name)
+      names.set(Number(c.release_id), set)
+    }
+    for (const c of candidates as { idx: number; id: string }[]) {
+      if (joined.has(c.idx)) continue
+      const have = names.get(Number(c.id)) ?? new Set<string>()
+      const incoming = decodeChunk(part[c.idx].payload).map((f) => f.name)
+      if (incoming.every((name) => !have.has(name))) {
+        joined.set(c.idx, Number(c.id))
+        for (const name of incoming) have.add(name)
+        names.set(Number(c.id), have)
+      }
+    }
+  }
+
   /**
    * Release aggregates in the catalog, one write-once chunk per release, and
    * probe candidates (a batch's, or files the fragment merger completed).
@@ -137,6 +186,7 @@ class BatchStore {
         [JSON.stringify(windows), groupId, UPLOAD_GAP_MS, UPLOAD_GAP_MS]
       )
       const joined = new Map<number, number>(matches.map((m: { idx: number; id: string }) => [m.idx, Number(m.id)]))
+      await this.joinAcrossGroups(trx, part, windows, groupId, joined)
       const row = (r: ReleaseChunk) => ({
         name: r.name,
         source: r.source,

tests/functional/batch_store.spec.ts

@@ -128,6 +128,14 @@ test.group('BatchStore', (group) => {
     assert.equal(rows[0].identity_key, rows[1].identity_key)
   })
 
+  test('an upload spread over groups (other files in another group) is one release', async ({ assert }) => {
+    await storeIn(GROUP, 1, upload(10, 'Some.Album-WEB-2026-FTD.part1.rar', 2, 'Some.Album-WEB-2026-FTD [2/3]', '2026-10-05T10:00:00Z'))
+    await storeIn('alt.binaries.other', 1, upload(10, 'Some.Album-WEB-2026-FTD.part2.rar', 2, 'Some.Album-WEB-2026-FTD [3/3]', '2026-10-05T10:00:30Z', 'alt.binaries.other'))
+    const rows = await releasesOf()
+    assert.lengthOf(rows, 1)
+    assert.equal(rows[0].files, 2)
+  })
+
   test('a release key taken meanwhile by another group\'s writer is skipped for the next number', async ({ assert }) => {
     await storeIn(GROUP, 1, upload(10, 'Some.Show.S01E01.1080p.mkv', 2, 'Some.Show.S01E01.1080p [1/1]', '2026-10-05T10:00:00Z'))
     const [first] = await releasesOf()