Идентичность на ULID: download_infohash, guarded-дедуп, миграция (ulid-identity)

Все сущности переехали с INTEGER AUTOINCREMENT на TEXT ULID (lowercase, internal/ident — единая точка генерации и разбора; oklog/ulid). Инфохэши загрузки — множество (download_infohash, v1/v2 гибридных торрентов): дедуп и сопоставление в поллинге по любому из хешей, magnet-парсер отдаёт оба хеша гибридной ссылки, усечённый v2-хеш v2-only раздач не хранится. Инвариант «не более одной активной загрузки на infohash» вместо снятого unique-индекса держат guarded-методы store в одной write-транзакции (_txlock=immediate): CreateDownloadIfNoActive (приём/adopt, с доносом недостающих хешей), ActivateIfNoOtherActive (retry/recovery/relink, отказ до побочных эффектов), guarded AddInfohashes; SetDownloadState отклоняет терминал→активное как механический бэкстоп. Миграция 0006 — первая Go-миграция goose: пересоздание таблиц при включённых FK, backfill ULID с timestamp из created_at (хронология id сохранена), разнос infohash, удаление idempotency_key. BREAKING: формат id в URL/логах/Telegram, REST-поля id (string) и infohashes (список). Новая конвенция docs/conventions/database.md (без числовых PK), корреляция в логах grep'ом по голому ULID, ER-схема обновлена. Спеки: новая capability identity, MODIFIED в state-reconciliation; change заархивирован. Пройдены ревью дизайна и кода (по 8 углов), все находки исправлены с регрессионными тестами. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-07-02 21:25:00 +03:00
parent b808ceff25
commit 37f2f6481a
53 changed files with 3640 additions and 1035 deletions
@@ -24,9 +24,17 @@ const errCodeQbitAdd = "qbit_add"

 // Store — нужная ingest часть хранилища.
 type Store interface {
-	FindActiveByInfohash(ctx context.Context, infohash string) (*store.Download, error)
-	CreateDownload(ctx context.Context, d *store.Download) (int64, error)
-	SetDownloadState(ctx context.Context, id int64, state store.State, errCode, errMsg string) error
+	// FindActiveByInfohash — быстрый читающий дедуп-чек (до вызова LLM-namer);
+	// авторитетная проверка — внутри CreateDownloadIfNoActive.
+	FindActiveByInfohash(ctx context.Context, hashes ...string) (*store.Download, error)
+	// CreateDownloadIfNoActive атомарно проверяет инвариант «одна активная
+	// загрузка на infohash» и заводит задачу; вернувшаяся existing ≠ nil —
+	// дедуп на активную задачу (недостающие хеши вызова метод доносит сам).
+	CreateDownloadIfNoActive(ctx context.Context, d *store.Download, hashes []string) (*store.Download, error)
+	// AddInfohashes доносит задаче недостающие хеши (guarded). Нужен на
+	// быстром дедуп-пути, который не доходит до CreateDownloadIfNoActive.
+	AddInfohashes(ctx context.Context, downloadID string, hashes []string) error
+	SetDownloadState(ctx context.Context, id string, state store.State, errCode, errMsg string) error
 }

 // QBittorrent — нужная ingest часть клиента qBittorrent.
@@ -58,7 +66,7 @@ type Service struct {
 	// не удалось). Closure, а не worker.Notifier: приёмное падение в qBit не
 	// попадает в поллинг-цикл worker (раздачи нет), поэтому уведомляет ingest
 	// сам; closure избавляет ядро приёма от зависимости на пакет worker.
-	notifyFailed func(downloadID int64)
+	notifyFailed func(downloadID string)
 }

 // New собирает сервис приёма. namer опционален (nil → отображаемое имя не
@@ -68,7 +76,7 @@ func New(st Store, qb QBittorrent, namer Namer, cfg Config, log *slog.Logger) *S
 }

 // SetFailureNotifier подключает пинг о падении приёма (до начала работы).
-func (s *Service) SetFailureNotifier(fn func(downloadID int64)) { s.notifyFailed = fn }
+func (s *Service) SetFailureNotifier(fn func(downloadID string)) { s.notifyFailed = fn }

 // Request — входной запрос приёма.
 type Request struct {
@@ -78,8 +86,8 @@ type Request struct {

 // Result — итог приёма.
 type Result struct {
-	DownloadID   int64
-	Infohash     string
+	DownloadID   string
+	Infohashes   []string // все хеши источника (гибридный magnet: v1 и v2, v1 первым)
 	State        store.State
 	Deduplicated bool // присоединились к уже активной задаче, нового добавления не было
 }
@@ -100,16 +108,14 @@ func (s *Service) Ingest(ctx context.Context, req Request) (Result, error) {
 	log := s.log.With("capability", capIngest, "infohash", info.Infohash)
 	ctx = logctx.With(ctx, log)

-	if existing, err := s.store.FindActiveByInfohash(ctx, info.Infohash); err != nil {
+	// Быстрый дедуп-чек до дорогого LLM-namer; авторитетная (атомарная)
+	// проверка — внутри CreateDownloadIfNoActive ниже. Дедуп — по ЛЮБОМУ из
+	// хешей источника: гибридный magnet несёт и v1, и v2.
+	if existing, err := s.store.FindActiveByInfohash(ctx, info.Infohashes...); err != nil {
 		return Result{}, fmt.Errorf("ingest: lookup active: %w", err)
 	} else if existing != nil {
 		log.Info("download attached to active", "download_id", existing.ID, "state", existing.State)
-		return Result{
-			DownloadID:   existing.ID,
-			Infohash:     info.Infohash,
-			State:        existing.State,
-			Deduplicated: true,
-		}, nil
+		return s.attached(ctx, info, existing), nil
 	}

 	// Отображаемое имя для списка qBit — best-effort: не валит приём.
@@ -122,18 +128,30 @@ func (s *Service) Ingest(ctx context.Context, req Request) (Result, error) {
 	}

 	d := &store.Download{
-		SourceType:     store.SourceMagnet,
-		SourceRef:      source,
-		DisplayName:    rename, // то же имя, что уходит в qBittorrent (rename); заголовок в веб-UI
-		Context:        req.Context,
-		Infohash:       store.NullString(info.Infohash),
-		IdempotencyKey: store.NullString(info.Infohash),
-		State:          store.StateDownloading,
+		SourceType:  store.SourceMagnet,
+		SourceRef:   source,
+		DisplayName: rename, // то же имя, что уходит в qBittorrent (rename); заголовок в веб-UI
+		Context:     req.Context,
+		State:       store.StateDownloading,
 	}
-	id, err := s.store.CreateDownload(ctx, d)
+	// Все хеши из magnet (гибридный несёт v1 и v2); kind store выведет по длине.
+	existing, err := s.store.CreateDownloadIfNoActive(ctx, d, info.Infohashes)
 	if err != nil {
 		return Result{}, fmt.Errorf("ingest: create download: %w", err)
 	}
+	if existing != nil {
+		// Гонка с параллельным приёмом/discover: активная задача появилась
+		// после быстрого чека — присоединяемся к ней (хеши донёс сам
+		// CreateDownloadIfNoActive).
+		log.Info("download attached to active", "download_id", existing.ID, "state", existing.State)
+		return Result{
+			DownloadID:   existing.ID,
+			Infohashes:   info.Infohashes,
+			State:        existing.State,
+			Deduplicated: true,
+		}, nil
+	}
+	id := d.ID
 	log = log.With("download_id", id)
 	ctx = logctx.With(ctx, log)

@@ -156,14 +174,33 @@ func (s *Service) Ingest(ctx context.Context, req Request) (Result, error) {
 			// сами, чтобы приёмные провалы тоже доходили до автора.
 			go s.notifyFailed(id)
 		}
-		return Result{DownloadID: id, Infohash: info.Infohash, State: store.StateFailed},
+		return Result{DownloadID: id, Infohashes: info.Infohashes, State: store.StateFailed},
 			fmt.Errorf("ingest: add to qbittorrent: %w", addErr)
 	}

 	log.Info("download accepted", "category", s.cfg.Category)
 	return Result{
 		DownloadID: id,
-		Infohash:   info.Infohash,
+		Infohashes: info.Infohashes,
 		State:      store.StateDownloading,
 	}, nil
 }
+
+// attached — итог дедупа на быстром чеке: присоединились к уже активной
+// задаче и доносим ей недостающие хеши источника (гибридный magnet мог
+// принести хеш, которого задача ещё не знает; guarded-путь через
+// CreateDownloadIfNoActive сюда не доходит). Донос — best-effort: конфликт
+// хеша с другой активной задачей логируется, приём не валится.
+func (s *Service) attached(ctx context.Context, info magnet.Info, existing *store.Download) Result {
+	if len(info.Infohashes) > len(existing.Infohashes) {
+		if err := s.store.AddInfohashes(ctx, existing.ID, info.Infohashes); err != nil {
+			logctx.FromOr(ctx, s.log).Warn("ingest top-up infohashes failed", "error", err)
+		}
+	}
+	return Result{
+		DownloadID:   existing.ID,
+		Infohashes:   info.Infohashes,
+		State:        existing.State,
+		Deduplicated: true,
+	}
+}
@@ -8,6 +8,7 @@ import (
 	"testing"
 	"time"

+	"git.vakhrushev.me/av/jellybit/internal/ident"
 	"git.vakhrushev.me/av/jellybit/internal/qbt"
 	"git.vakhrushev.me/av/jellybit/internal/store"
 )
@@ -19,29 +20,39 @@ const sampleInfohash = "541adcff3b6dd5dba7088ea83317d9d6fac331d6"
 type fakeStore struct {
 	active     *store.Download
 	created    []store.Download
-	nextID     int64
+	hashes     [][]string
+	toppedUp   []string
 	stateCalls []stateCall
 }

 type stateCall struct {
-	id    int64
+	id    string
 	state store.State
 	code  string
 	msg   string
 }

-func (f *fakeStore) FindActiveByInfohash(_ context.Context, _ string) (*store.Download, error) {
+func (f *fakeStore) FindActiveByInfohash(_ context.Context, _ ...string) (*store.Download, error) {
 	return f.active, nil
 }

-func (f *fakeStore) CreateDownload(_ context.Context, d *store.Download) (int64, error) {
-	f.nextID++
-	d.ID = f.nextID
+func (f *fakeStore) CreateDownloadIfNoActive(_ context.Context, d *store.Download, hashes []string) (*store.Download, error) {
+	if f.active != nil {
+		return f.active, nil
+	}
+	d.ID = ident.NewID()
 	f.created = append(f.created, *d)
-	return f.nextID, nil
+	f.hashes = append(f.hashes, hashes)
+	return nil, nil
 }

-func (f *fakeStore) SetDownloadState(_ context.Context, id int64, st store.State, code, msg string) error {
+func (f *fakeStore) AddInfohashes(_ context.Context, id string, hashes []string) error {
+	f.toppedUp = append(f.toppedUp, hashes...)
+	_ = id
+	return nil
+}
+
+func (f *fakeStore) SetDownloadState(_ context.Context, id string, st store.State, code, msg string) error {
 	f.stateCalls = append(f.stateCalls, stateCall{id, st, code, msg})
 	return nil
 }
@@ -90,8 +101,8 @@ func TestIngestHappyPath(t *testing.T) {
 	if err != nil {
 		t.Fatalf("Ingest: %v", err)
 	}
-	if res.Infohash != sampleInfohash {
-		t.Errorf("infohash = %q", res.Infohash)
+	if len(res.Infohashes) != 1 || res.Infohashes[0] != sampleInfohash {
+		t.Errorf("infohashes = %v", res.Infohashes)
 	}
 	if res.State != store.StateDownloading || res.Deduplicated {
 		t.Errorf("res = %+v", res)
@@ -99,9 +110,12 @@ func TestIngestHappyPath(t *testing.T) {
 	if len(fs.created) != 1 {
 		t.Fatalf("создано задач: %d, want 1", len(fs.created))
 	}
-	if got := fs.created[0]; got.Context != "Дюна 2" || got.Infohash.String != sampleInfohash {
+	if got := fs.created[0]; got.Context != "Дюна 2" {
 		t.Errorf("сохранённая задача: %+v", got)
 	}
+	if len(fs.hashes) != 1 || len(fs.hashes[0]) != 1 || fs.hashes[0][0] != sampleInfohash {
+		t.Errorf("хеши задачи: %v", fs.hashes)
+	}
 	if len(fq.added) != 1 {
 		t.Fatalf("вызовов qbt.Add: %d, want 1", len(fq.added))
 	}
@@ -149,15 +163,15 @@ func TestIngestEmptyNameOmitsRename(t *testing.T) {
 }

 func TestIngestIdempotent(t *testing.T) {
-	existing := &store.Download{ID: 7, State: store.StateDownloading}
+	existing := &store.Download{ID: "01hzzzexisting000000000000", State: store.StateDownloading}
 	fs := &fakeStore{active: existing}
 	fq := &fakeQbt{}
 	res, err := newService(fs, fq).Ingest(context.Background(), Request{Source: sampleMagnet})
 	if err != nil {
 		t.Fatalf("Ingest: %v", err)
 	}
-	if !res.Deduplicated || res.DownloadID != 7 {
-		t.Errorf("ожидалось присоединение к задаче 7: %+v", res)
+	if !res.Deduplicated || res.DownloadID != existing.ID {
+		t.Errorf("ожидалось присоединение к существующей задаче: %+v", res)
 	}
 	if len(fs.created) != 0 {
 		t.Error("не должно создаваться новой задачи")
@@ -167,6 +181,29 @@ func TestIngestIdempotent(t *testing.T) {
 	}
 }

+// Быстрый дедуп-путь доносит существующей задаче недостающие хеши
+// гибридного magnet (иначе последующий приём по второму хешу создал бы
+// вторую активную задачу).
+func TestIngestDedupTopsUpHashes(t *testing.T) {
+	const v2 = "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef"
+	existing := &store.Download{
+		ID: "01hzzzexisting000000000000", State: store.StateDownloading,
+		Infohashes: []store.Infohash{{DownloadID: "01hzzzexisting000000000000", Infohash: sampleInfohash, Kind: store.HashV1}},
+	}
+	fs := &fakeStore{active: existing}
+	res, err := newService(fs, &fakeQbt{}).Ingest(context.Background(),
+		Request{Source: sampleMagnet + "&xt=urn:btmh:1220" + v2})
+	if err != nil {
+		t.Fatalf("Ingest: %v", err)
+	}
+	if !res.Deduplicated {
+		t.Fatalf("ожидался дедуп: %+v", res)
+	}
+	if len(fs.toppedUp) != 2 {
+		t.Errorf("хеши не донесены существующей задаче: %v", fs.toppedUp)
+	}
+}
+
 func TestIngestQbitErrorMarksFailed(t *testing.T) {
 	fs := &fakeStore{}
 	fq := &fakeQbt{err: errors.New("connection refused")}
@@ -186,16 +223,16 @@ func TestIngestQbitErrorNotifies(t *testing.T) {
 	fs := &fakeStore{}
 	fq := &fakeQbt{err: errors.New("connection refused")}
 	svc := newService(fs, fq)
-	got := make(chan int64, 1)
-	svc.SetFailureNotifier(func(id int64) { got <- id })
+	got := make(chan string, 1)
+	svc.SetFailureNotifier(func(id string) { got <- id })

 	if _, err := svc.Ingest(context.Background(), Request{Source: sampleMagnet}); err == nil {
 		t.Fatal("ожидалась ошибка")
 	}
 	select {
 	case id := <-got:
-		if id == 0 {
-			t.Errorf("уведомление с нулевым id")
+		if id == "" {
+			t.Errorf("уведомление с пустым id")
 		}
 	case <-time.After(2 * time.Second):
 		t.Fatal("уведомление о падении приёма не пришло")