transcriber/internal/adapter/repo/pocketbase/transcript_job_repo_test.go

package pocketbase

import (
	"net/http"
	"net/http/httptest"
	"strings"
	"sync"
	"testing"
	"time"

	"github.com/pocketbase/pocketbase/apis"
	"github.com/pocketbase/pocketbase/core"
	"github.com/pocketbase/pocketbase/tools/types"
	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"

	"git.vakhrushev.me/av/transcriber/internal/contract"
	"git.vakhrushev.me/av/transcriber/internal/entity"

	"git.vakhrushev.me/av/transcriber/internal/adapter/repo/pocketbase/migrations"
)

// newTestApp поднимает хранилище на пустом каталоге и накатывает схему — тем же
// путём, каким это делает сервис при старте.
func newTestApp(t *testing.T) core.App {
	t.Helper()

	app, err := New(t.TempDir())
	require.NoError(t, err)
	t.Cleanup(func() {
		if err := app.ResetBootstrapState(); err != nil {
			t.Logf("не удалось закрыть хранилище: %v", err)
		}
	})

	return app
}

// newFile заводит запись о файле: ссылка на неё у задачи обязательна схемой.
func newFile(t *testing.T, app core.App) *entity.File {
	t.Helper()

	repo := NewFileRepository(app)
	work, err := repo.Stage(".mp3", strings.NewReader("запись"))
	require.NoError(t, err)
	defer func() { require.NoError(t, work.Close()) }()

	file, err := repo.CreateLocal("sample.mp3", work)
	require.NoError(t, err)
	return file
}

func newJob(t *testing.T, repo *TranscriptJobRepository, state string) *entity.TranscribeJob {
	t.Helper()

	file := newFile(t, repo.app)
	job := &entity.TranscribeJob{State: state, Source: entity.SourceApi, FileID: &file.Id}
	require.NoError(t, repo.Create(job))
	return job
}

// Захват неделим: выбор подходящей задачи и пометка её захваченной идут вместе.
// Двум вызывающим, пришедшим за одним состоянием разом, запись достаётся
// одному — на этом стоит инвариант «Принятая запись не теряется молча».
func TestFindAndAcquire_OnlyOneOfThreeGetsTheJob(t *testing.T) {
	app := newTestApp(t)
	repo := NewTranscriptJobRepository(app)

	job := newJob(t, repo, entity.StateCreated)

	const racers = 3

	var (
		wg       sync.WaitGroup
		mu       sync.Mutex
		got      []*entity.TranscribeJob
		notFound int
	)

	start := make(chan struct{})
	for i := 0; i < racers; i++ {
		wg.Add(1)
		go func(n int) {
			defer wg.Done()
			<-start

			acquired, err := repo.FindAndAcquire(entity.StateCreated, "holder", time.Now().Add(-time.Hour))

			mu.Lock()
			defer mu.Unlock()
			if err != nil {
				var missing *contract.JobNotFoundError
				if assert.ErrorAs(t, err, &missing) {
					notFound++
				}
				return
			}
			got = append(got, acquired)
		}(i)
	}

	close(start)
	wg.Wait()

	require.Len(t, got, 1, "запись получает ровно один из трёх захватов")
	assert.Equal(t, job.Id, got[0].Id)
	assert.Equal(t, racers-1, notFound, "остальные получают признак «работы нет»")
}

// Захваченная задача второй раз не выдаётся, пока срок захвата не истёк.
func TestFindAndAcquire_AcquiredJobIsNotHandedOutAgain(t *testing.T) {
	app := newTestApp(t)
	repo := NewTranscriptJobRepository(app)

	newJob(t, repo, entity.StateCreated)

	first, err := repo.FindAndAcquire(entity.StateCreated, "first", time.Now().Add(-time.Hour))
	require.NoError(t, err)
	require.NotNil(t, first)

	_, err = repo.FindAndAcquire(entity.StateCreated, "second", time.Now().Add(-time.Hour))

	var missing *contract.JobNotFoundError
	assert.ErrorAs(t, err, &missing, "захваченная задача второму не выдаётся")
}

// Захват протухает, и задача достаётся снова. Время захвата кладётся **не**
// нашим кодом, а тем же путём, что и `created`: проверка, кладущая его своим
// форматом, была бы зелена и тогда, когда сравнение вида сломано.
func TestFindAndAcquire_RottenAcquisitionIsHandedOutAgain(t *testing.T) {
	app := newTestApp(t)
	repo := NewTranscriptJobRepository(app)

	job := newJob(t, repo, entity.StateCreated)

	_, err := repo.FindAndAcquire(entity.StateCreated, "first", time.Now().Add(-time.Hour))
	require.NoError(t, err)

	// Задним числом — записью коллекции, то есть тем же слоем, который пишет
	// собственные времена хранилища.
	record, err := app.FindRecordById(migrations.JobsCollection, job.Id)
	require.NoError(t, err)
	record.Set("acquire_time", types.NowDateTime().Add(-2*time.Hour))
	require.NoError(t, app.Save(record))

	again, err := repo.FindAndAcquire(entity.StateCreated, "second", time.Now().Add(-time.Hour))
	require.NoError(t, err, "протухший захват не мешает выдать задачу следующему")
	assert.Equal(t, job.Id, again.Id)
}

// Пауза держит задачу от выдачи, пока не кончится.
func TestFindAndAcquire_DelayedJobIsNotHandedOut(t *testing.T) {
	app := newTestApp(t)
	repo := NewTranscriptJobRepository(app)

	job := newJob(t, repo, entity.StateCreated)

	delay := time.Now().Add(time.Hour)
	job.DelayTime = &delay
	require.NoError(t, repo.Save(job, ""))

	_, err := repo.FindAndAcquire(entity.StateCreated, "holder", time.Now().Add(-time.Hour))

	var missing *contract.JobNotFoundError
	assert.ErrorAs(t, err, &missing, "задача не выдаётся, пока пауза не кончилась")
}

// Число попыток растёт при каждом захвате: только так попытка засчитывается и
// задаче, брошенной вместе с процессом.
func TestFindAndAcquire_AttemptsGrowOnEveryAcquisition(t *testing.T) {
	app := newTestApp(t)
	repo := NewTranscriptJobRepository(app)

	newJob(t, repo, entity.StateCreated)

	for expected := 1; expected <= 3; expected++ {
		acquired, err := repo.FindAndAcquire(entity.StateCreated, "holder", time.Now().Add(time.Hour))
		require.NoError(t, err)
		assert.Equal(t, expected, acquired.Attempts)
	}
}

// Захват отдаёт задачу целиком, а не только её ключ: сырой запрос идёт мимо
// записей коллекции, и расхождение перечня колонок иначе проявилось бы как
// потерянное поле.
func TestFindAndAcquire_ReturnsWholeJob(t *testing.T) {
	app := newTestApp(t)
	repo := NewTranscriptJobRepository(app)

	chatId := int64(4242)
	replyId := 17
	opId := "operation-id"
	text := "расшифровка"

	file := newFile(t, app)

	job := &entity.TranscribeJob{
		State:             entity.StateTranscribe,
		Source:            entity.SourceTelegram,
		FileID:            &file.Id,
		TgChatId:          &chatId,
		TgReplyMessageId:  &replyId,
		RecognitionOpID:   &opId,
		TranscriptionText: &text,
	}
	require.NoError(t, repo.Create(job))

	acquired, err := repo.FindAndAcquire(entity.StateTranscribe, "holder", time.Now().Add(-time.Hour))
	require.NoError(t, err)

	assert.Equal(t, job.Id, acquired.Id)
	assert.Equal(t, entity.StateTranscribe, acquired.State)
	assert.Equal(t, entity.SourceTelegram, acquired.Source)
	require.NotNil(t, acquired.TgChatId)
	assert.Equal(t, chatId, *acquired.TgChatId)
	require.NotNil(t, acquired.TgReplyMessageId)
	assert.Equal(t, replyId, *acquired.TgReplyMessageId)
	require.NotNil(t, acquired.RecognitionOpID)
	assert.Equal(t, opId, *acquired.RecognitionOpID)
	require.NotNil(t, acquired.TranscriptionText)
	assert.Equal(t, text, *acquired.TranscriptionText)
	assert.False(t, acquired.CreatedAt.IsZero(), "время заведения доехало")
}

// Шаг, потерявший захват за время работы, результата не пишет: иначе два
// воркера пишут в одну задачу по очереди, а отправитель получает два ответа.
func TestSave_RefusesWriteFromLostAcquisition(t *testing.T) {
	app := newTestApp(t)
	repo := NewTranscriptJobRepository(app)

	newJob(t, repo, entity.StateCreated)

	mine, err := repo.FindAndAcquire(entity.StateCreated, "mine", time.Now().Add(-time.Hour))
	require.NoError(t, err)

	// Задача досталась другому, пока шаг работал.
	record, err := app.FindRecordById(migrations.JobsCollection, mine.Id)
	require.NoError(t, err)
	record.Set("acquisition_id", "someone-else")
	require.NoError(t, app.Save(record))

	mine.MoveToState(entity.StateConverted)
	err = repo.Save(mine, "mine")

	var lost *contract.LostAcquisitionError
	require.ErrorAs(t, err, &lost)

	// И состояние не поехало.
	after, err := repo.GetByID(mine.Id)
	require.NoError(t, err)
	assert.Equal(t, entity.StateCreated, after.State)
}

// Пустой держатель значит «задача не захватывалась» — так её сохраняет приём.
func TestSave_WithoutHolderWritesAnyway(t *testing.T) {
	app := newTestApp(t)
	repo := NewTranscriptJobRepository(app)

	job := newJob(t, repo, entity.StateCreated)
	job.MoveToState(entity.StateConverted)

	require.NoError(t, repo.Save(job, ""))

	after, err := repo.GetByID(job.Id)
	require.NoError(t, err)
	assert.Equal(t, entity.StateConverted, after.State)
}

// Правка состояния **запросом** — то есть из панели — чистит служебные поля
// прошлого состояния: те же, что чистит переход из кода. Иначе владелец,
// вернувший мёртвую задачу в работу, получил бы задачу, которая не выдаётся
// захвату и умирает от первого же отказа, и не узнал бы об этом.
func TestPanelRules_StateChangeByRequestClearsAcquisition(t *testing.T) {
	app := newTestApp(t)
	BindPanelRules(app)

	repo := NewTranscriptJobRepository(app)
	job := newJob(t, repo, entity.StateCreated)

	acquired, err := repo.FindAndAcquire(entity.StateCreated, "holder", time.Now().Add(-time.Hour))
	require.NoError(t, err)
	require.NotNil(t, acquired.AcquisitionID)

	record, err := app.FindRecordById(migrations.JobsCollection, job.Id)
	require.NoError(t, err)
	record.Set("attempts", 5)
	record.Set("state", entity.StateDead)
	require.NoError(t, app.Save(record))

	// Владелец возвращает задачу в работу правкой состояния в панели — то есть
	// запросом к записи, а не сохранением из кода.
	patchRecord(t, app, job.Id, `{"state":"`+entity.StateCreated+`"}`)

	after, err := repo.GetByID(job.Id)
	require.NoError(t, err)
	assert.Nil(t, after.AcquisitionID, "признак захвата снят")
	assert.Nil(t, after.AcquireTime, "время захвата снято")
	assert.Nil(t, after.DelayTime, "пауза снята")
	assert.Equal(t, 0, after.Attempts, "число попыток обнулено")

	// И ближайший захват задачу выдаёт.
	again, err := repo.FindAndAcquire(entity.StateCreated, "next", time.Now().Add(-time.Hour))
	require.NoError(t, err)
	assert.Equal(t, job.Id, again.Id)
}

// Обратная сторона того же правила, и она дороже: правила панели MUST не
// трогать записи, которые правит сам конвейер. Модельный хук их не различал, и
// пауза, поставленная шагом вместе со сменой состояния, стиралась тем же
// сохранением, а число попыток мёртвой задачи приходило владельцу нулём.
func TestPanelRules_DoNotTouchPipelineWrites(t *testing.T) {
	app := newTestApp(t)
	BindPanelRules(app)

	repo := NewTranscriptJobRepository(app)
	job := newJob(t, repo, entity.StateConverted)

	acquired, err := repo.FindAndAcquire(entity.StateConverted, "holder", time.Now().Add(-time.Hour))
	require.NoError(t, err)

	// Шаг ставит задержку опроса вместе со сменой состояния.
	delay := time.Now().Add(10 * time.Second)
	acquired.MoveToStateAndDelay(entity.StateTranscribe, &delay)
	require.NoError(t, repo.Save(acquired, "holder"))

	after, err := repo.GetByID(job.Id)
	require.NoError(t, err)
	require.NotNil(t, after.DelayTime, "задержка, поставленная шагом, пережила сохранение")

	// Переход в «мертва» хранит число попыток намеренно: по нему владелец видит,
	// сколько раз мы пробовали.
	after.Attempts = 6
	after.Die("attempts exhausted: 6")
	require.NoError(t, repo.Save(after, ""))

	dead, err := repo.GetByID(job.Id)
	require.NoError(t, err)
	assert.Equal(t, entity.StateDead, dead.State)
	assert.Equal(t, 6, dead.Attempts, "число попыток мёртвой задачи сохранено")
}

// patchRecord правит запись тем же путём, каким её правит панель: запросом к
// API от имени владельца.
func patchRecord(t *testing.T, app core.App, recordID, body string) {
	t.Helper()

	superusers, err := app.FindCollectionByNameOrId(core.CollectionNameSuperusers)
	require.NoError(t, err)

	owner := core.NewRecord(superusers)
	owner.Set("email", "owner@example.com")
	owner.Set("password", "ownerpassword123")
	require.NoError(t, app.Save(owner))

	token, err := owner.NewStaticAuthToken(time.Hour)
	require.NoError(t, err)

	router, err := apis.NewRouter(app)
	require.NoError(t, err)
	mux, err := router.BuildMux()
	require.NoError(t, err)

	req := httptest.NewRequest(
		http.MethodPatch,
		"/api/collections/"+migrations.JobsCollection+"/records/"+recordID,
		strings.NewReader(body),
	)
	req.Header.Set("Content-Type", "application/json")
	req.Header.Set("Authorization", token)

	w := httptest.NewRecorder()
	mux.ServeHTTP(w, req)
	require.Equal(t, http.StatusOK, w.Code, "правка записи владельцем: %s", w.Body.String())
}

// Правка владельца в панели переживает сохранение шага. Шаг держит задачу
// снимком с момента захвата и до своего сохранения — до восьми часов, — и
// безусловная запись снимка стёрла бы правку молча: ни строки в журнале, ни
// отказа в панели.
func TestSave_KeepsOwnerEditMadeWhileStepHeldTheJob(t *testing.T) {
	app := newTestApp(t)
	BindPanelRules(app)

	repo := NewTranscriptJobRepository(app)

	file := newFile(t, app)
	chatId := int64(111)
	job := &entity.TranscribeJob{
		State:    entity.StateCreated,
		Source:   entity.SourceTelegram,
		FileID:   &file.Id,
		TgChatId: &chatId,
	}
	require.NoError(t, repo.Create(job))

	// Шаг захватил задачу и работает.
	acquired, err := repo.FindAndAcquire(entity.StateCreated, "holder", time.Now().Add(-time.Hour))
	require.NoError(t, err)

	// Владелец правит в панели поле, которого конвейер не касается.
	patchRecord(t, app, job.Id, `{"tg_chat_id":999999}`)

	// Шаг доработал и сохраняет свой снимок.
	acquired.MoveToState(entity.StateConverted)
	require.NoError(t, repo.Save(acquired, "holder"))

	after, err := repo.GetByID(job.Id)
	require.NoError(t, err)
	assert.Equal(t, entity.StateConverted, after.State, "шаг свой результат записал")
	require.NotNil(t, after.TgChatId)
	assert.Equal(t, int64(999999), *after.TgChatId, "правка владельца пережила сохранение шага")
}