番組登録を追加（登録時の過去回の選択、一時停止・再開・解除、既知の回の記録）
This commit is contained in:
@@ -0,0 +1,8 @@
|
||||
CREATE TABLE known_episodes (
|
||||
series_id TEXT NOT NULL REFERENCES subscriptions(series_id) ON DELETE CASCADE,
|
||||
episode_id TEXT NOT NULL,
|
||||
first_seen_at TIMESTAMPTZ NOT NULL DEFAULT now(),
|
||||
PRIMARY KEY (series_id, episode_id)
|
||||
);
|
||||
|
||||
CREATE INDEX download_jobs_series_idx ON download_jobs (series_id);
|
||||
+5
-10
@@ -59,6 +59,7 @@ type Subscription struct {
|
||||
Enabled bool
|
||||
CreatedAt time.Time
|
||||
LastCheckedAt *time.Time
|
||||
JobCount int64
|
||||
}
|
||||
|
||||
const jobColumns = `id, episode_id, series_id, series_title, title, broadcaster_name,
|
||||
@@ -246,21 +247,15 @@ func (s *Store) EpisodeStatuses(ctx context.Context, ids []string) (map[string]s
|
||||
return result, rows.Err()
|
||||
}
|
||||
|
||||
// AddSubscription は登録済みシリーズのタイトルと有効状態を変更しない。
|
||||
func (s *Store) AddSubscription(ctx context.Context, seriesID, title string) (bool, error) {
|
||||
tag, err := s.pool.Exec(ctx, `INSERT INTO subscriptions (series_id, series_title)
|
||||
VALUES ($1,$2) ON CONFLICT (series_id) DO NOTHING`, seriesID, title)
|
||||
return tag.RowsAffected() == 1, err
|
||||
}
|
||||
|
||||
func (s *Store) DeleteSubscription(ctx context.Context, seriesID string) (bool, error) {
|
||||
tag, err := s.pool.Exec(ctx, `DELETE FROM subscriptions WHERE series_id=$1`, seriesID)
|
||||
return tag.RowsAffected() == 1, err
|
||||
}
|
||||
|
||||
func (s *Store) ListSubscriptions(ctx context.Context) ([]Subscription, error) {
|
||||
rows, err := s.pool.Query(ctx, `SELECT series_id, series_title, enabled, created_at, last_checked_at
|
||||
FROM subscriptions ORDER BY created_at DESC, series_id`)
|
||||
rows, err := s.pool.Query(ctx, `SELECT s.series_id, s.series_title, s.enabled, s.created_at, s.last_checked_at,
|
||||
(SELECT count(*) FROM download_jobs j WHERE j.series_id=s.series_id)
|
||||
FROM subscriptions s ORDER BY s.created_at DESC, s.series_id`)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
@@ -268,7 +263,7 @@ func (s *Store) ListSubscriptions(ctx context.Context) ([]Subscription, error) {
|
||||
var subs []Subscription
|
||||
for rows.Next() {
|
||||
var sub Subscription
|
||||
if err := rows.Scan(&sub.SeriesID, &sub.SeriesTitle, &sub.Enabled, &sub.CreatedAt, &sub.LastCheckedAt); err != nil {
|
||||
if err := rows.Scan(&sub.SeriesID, &sub.SeriesTitle, &sub.Enabled, &sub.CreatedAt, &sub.LastCheckedAt, &sub.JobCount); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
subs = append(subs, sub)
|
||||
|
||||
@@ -113,7 +113,7 @@ func TestMigrate(t *testing.T) {
|
||||
if err := pool.QueryRow(ctx, `SELECT count(*), min(applied_at) FROM schema_migrations`).Scan(&count, &second); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if count != 1 || !first.Equal(second) {
|
||||
if count != 2 || !first.Equal(second) {
|
||||
t.Fatalf("再適用で履歴が変わった: %d %v %v", count, first, second)
|
||||
}
|
||||
}
|
||||
@@ -138,11 +138,48 @@ func TestMigrateConcurrent(t *testing.T) {
|
||||
if err := pool.QueryRow(context.Background(), `SELECT count(*) FROM schema_migrations`).Scan(&count); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if count != 1 {
|
||||
if count != 2 {
|
||||
t.Fatalf("適用件数: %d", count)
|
||||
}
|
||||
}
|
||||
|
||||
func TestMigrateAddsKnownEpisodesToExistingDatabase(t *testing.T) {
|
||||
pool := testPool(t, false)
|
||||
ctx := context.Background()
|
||||
initial, err := migrations.ReadFile("migrations/001_initial.sql")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := pool.Exec(ctx, string(initial), pgx.QueryExecModeSimpleProtocol); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := pool.Exec(ctx, `CREATE TABLE schema_migrations (version TEXT PRIMARY KEY, applied_at TIMESTAMPTZ NOT NULL DEFAULT now())`); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := pool.Exec(ctx, `INSERT INTO schema_migrations (version) VALUES ('migrations/001_initial.sql')`); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := pool.Exec(ctx, `INSERT INTO subscriptions (series_id, series_title) VALUES ('existing', '既存番組')`); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := Migrate(ctx, pool); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var count int
|
||||
if err := pool.QueryRow(ctx, `SELECT count(*) FROM schema_migrations`).Scan(&count); err != nil || count != 2 {
|
||||
t.Fatalf("移行履歴: %d %v", count, err)
|
||||
}
|
||||
if _, err := pool.Exec(ctx, `INSERT INTO known_episodes (series_id, episode_id) VALUES ('existing', 'ep1')`); err != nil {
|
||||
t.Fatalf("既存番組への既知記録: %v", err)
|
||||
}
|
||||
if _, err := pool.Exec(ctx, `DELETE FROM subscriptions WHERE series_id='existing'`); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := pool.QueryRow(ctx, `SELECT count(*) FROM known_episodes WHERE series_id='existing'`).Scan(&count); err != nil || count != 0 {
|
||||
t.Fatalf("アップグレード後の cascade: %d %v", count, err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestDownloadJobIndexes(t *testing.T) {
|
||||
pool := testPool(t, true)
|
||||
rows, err := pool.Query(context.Background(), `SELECT indexname, indexdef FROM pg_indexes WHERE schemaname=current_schema() AND tablename='download_jobs'`)
|
||||
@@ -617,11 +654,11 @@ func TestListJobsFiltersAndLimit(t *testing.T) {
|
||||
func TestSubscriptions(t *testing.T) {
|
||||
s := New(testPool(t, true))
|
||||
ctx := context.Background()
|
||||
created, err := s.AddSubscription(ctx, "series", "番組")
|
||||
created, err := s.RegisterSubscription(ctx, "series", "番組", nil, false)
|
||||
if err != nil || !created {
|
||||
t.Fatalf("add: %v %v", created, err)
|
||||
}
|
||||
created, err = s.AddSubscription(ctx, "series", "別タイトル")
|
||||
created, err = s.RegisterSubscription(ctx, "series", "別タイトル", nil, false)
|
||||
if err != nil || created {
|
||||
t.Fatalf("duplicate: %v %v", created, err)
|
||||
}
|
||||
|
||||
@@ -0,0 +1,154 @@
|
||||
package store
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
|
||||
"github.com/jackc/pgx/v5"
|
||||
)
|
||||
|
||||
var ErrSubscriptionPaused = errors.New("subscription is paused")
|
||||
|
||||
// RegisterSubscription は登録と公開中の回の記録・任意の投入を一つの取引で行う。
|
||||
func (s *Store) RegisterSubscription(ctx context.Context, seriesID, title string, episodes []Episode, downloadExisting bool) (bool, error) {
|
||||
if seriesID == "" || title == "" {
|
||||
return false, errors.New("series id and title are required")
|
||||
}
|
||||
tx, err := s.pool.Begin(ctx)
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
defer tx.Rollback(ctx)
|
||||
var inserted string
|
||||
err = tx.QueryRow(ctx, `INSERT INTO subscriptions (series_id, series_title, last_checked_at)
|
||||
VALUES ($1,$2,now()) ON CONFLICT (series_id) DO NOTHING RETURNING series_id`, seriesID, title).Scan(&inserted)
|
||||
if errors.Is(err, pgx.ErrNoRows) {
|
||||
return false, nil
|
||||
}
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
if _, err := addKnownAndMaybeQueue(ctx, tx, seriesID, episodes, downloadExisting); err != nil {
|
||||
return false, err
|
||||
}
|
||||
if err := tx.Commit(ctx); err != nil {
|
||||
return false, err
|
||||
}
|
||||
return true, nil
|
||||
}
|
||||
|
||||
// QueueUnseenEpisodes は未確認の回だけを既知に記録し、同じ取引で待ち行列へ入れる。
|
||||
func (s *Store) QueueUnseenEpisodes(ctx context.Context, seriesID string, episodes []Episode) (int, error) {
|
||||
tx, err := s.pool.Begin(ctx)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
defer tx.Rollback(ctx)
|
||||
var enabled bool
|
||||
err = tx.QueryRow(ctx, `SELECT enabled FROM subscriptions WHERE series_id=$1 FOR UPDATE`, seriesID).Scan(&enabled)
|
||||
if errors.Is(err, pgx.ErrNoRows) {
|
||||
return 0, ErrInvalidTransition
|
||||
}
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
if !enabled {
|
||||
return 0, ErrSubscriptionPaused
|
||||
}
|
||||
count, err := addKnownAndMaybeQueue(ctx, tx, seriesID, episodes, true)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
if err := tx.Commit(ctx); err != nil {
|
||||
return 0, err
|
||||
}
|
||||
return count, nil
|
||||
}
|
||||
|
||||
func addKnownAndMaybeQueue(ctx context.Context, tx pgx.Tx, seriesID string, episodes []Episode, queue bool) (int, error) {
|
||||
count := 0
|
||||
for _, ep := range episodes {
|
||||
if ep.ID == "" || ep.SeriesID != seriesID {
|
||||
return 0, fmt.Errorf("invalid episode identity for series %s", seriesID)
|
||||
}
|
||||
var inserted string
|
||||
err := tx.QueryRow(ctx, `INSERT INTO known_episodes (series_id, episode_id)
|
||||
VALUES ($1,$2) ON CONFLICT DO NOTHING RETURNING episode_id`, seriesID, ep.ID).Scan(&inserted)
|
||||
if errors.Is(err, pgx.ErrNoRows) {
|
||||
continue
|
||||
}
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
count++
|
||||
if queue {
|
||||
_, err := tx.Exec(ctx, `INSERT INTO download_jobs
|
||||
(episode_id, series_id, series_title, title, broadcaster_name, duration_seconds, end_at, thumbnail_path, source)
|
||||
VALUES ($1,$2,$3,$4,$5,$6,$7,$8,'subscription')
|
||||
ON CONFLICT (episode_id) DO NOTHING`, ep.ID, seriesID, ep.SeriesTitle, ep.Title,
|
||||
ep.BroadcasterName, ep.DurationSeconds, ep.EndAt, ep.ThumbnailPath)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
}
|
||||
}
|
||||
return count, nil
|
||||
}
|
||||
|
||||
// UnknownEpisodeIDs は指定シリーズの未確認のエピソード ID を入力順で返す。
|
||||
func (s *Store) UnknownEpisodeIDs(ctx context.Context, seriesID string, ids []string) ([]string, error) {
|
||||
if len(ids) == 0 {
|
||||
return []string{}, nil
|
||||
}
|
||||
rows, err := s.pool.Query(ctx, `SELECT episode_id FROM known_episodes WHERE series_id=$1 AND episode_id=ANY($2)`, seriesID, ids)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
known := make(map[string]bool)
|
||||
for rows.Next() {
|
||||
var id string
|
||||
if err := rows.Scan(&id); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
known[id] = true
|
||||
}
|
||||
if err := rows.Err(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
var unknown []string
|
||||
emitted := make(map[string]bool)
|
||||
for _, id := range ids {
|
||||
if !known[id] && !emitted[id] {
|
||||
unknown = append(unknown, id)
|
||||
emitted[id] = true
|
||||
}
|
||||
}
|
||||
return unknown, nil
|
||||
}
|
||||
|
||||
func (s *Store) SetSubscriptionEnabled(ctx context.Context, seriesID string, enabled bool) error {
|
||||
tag, err := s.pool.Exec(ctx, `UPDATE subscriptions SET enabled=$2 WHERE series_id=$1`, seriesID, enabled)
|
||||
return transitionResult(tag.RowsAffected(), err)
|
||||
}
|
||||
|
||||
func (s *Store) SubscribedSeries(ctx context.Context, ids []string) (map[string]bool, error) {
|
||||
result := make(map[string]bool)
|
||||
if len(ids) == 0 {
|
||||
return result, nil
|
||||
}
|
||||
rows, err := s.pool.Query(ctx, `SELECT series_id FROM subscriptions WHERE series_id=ANY($1)`, ids)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
for rows.Next() {
|
||||
var id string
|
||||
if err := rows.Scan(&id); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
result[id] = true
|
||||
}
|
||||
return result, rows.Err()
|
||||
}
|
||||
@@ -0,0 +1,190 @@
|
||||
package store
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"sync"
|
||||
"testing"
|
||||
)
|
||||
|
||||
func seriesEpisode(id, seriesID string) Episode {
|
||||
ep := sampleEpisode(id)
|
||||
ep.SeriesID = seriesID
|
||||
return ep
|
||||
}
|
||||
|
||||
func TestRegisterSubscriptionSnapshotAndRollback(t *testing.T) {
|
||||
pool := testPool(t, true)
|
||||
s := New(pool)
|
||||
ctx := context.Background()
|
||||
episodes := []Episode{seriesEpisode("ep1", "series-1"), seriesEpisode("ep2", "series-1")}
|
||||
created, err := s.RegisterSubscription(ctx, "series-1", "番組", episodes, false)
|
||||
if err != nil || !created {
|
||||
t.Fatalf("既存回をダウンロードしない登録: %t %v", created, err)
|
||||
}
|
||||
subs, err := s.ListSubscriptions(ctx)
|
||||
if err != nil || len(subs) != 1 || subs[0].LastCheckedAt == nil || subs[0].LastCheckedAt.Before(subs[0].CreatedAt) {
|
||||
t.Fatalf("登録時の最終確認日時: %+v %v", subs, err)
|
||||
}
|
||||
unknown, err := s.UnknownEpisodeIDs(ctx, "series-1", []string{"ep1", "ep2", "ep3"})
|
||||
if err != nil || len(unknown) != 1 || unknown[0] != "ep3" {
|
||||
t.Fatalf("既知の記録: %v %v", unknown, err)
|
||||
}
|
||||
var jobs int
|
||||
if err := pool.QueryRow(ctx, `SELECT count(*) FROM download_jobs WHERE series_id='series-1'`).Scan(&jobs); err != nil || jobs != 0 {
|
||||
t.Fatalf("既存回が投入された: %d %v", jobs, err)
|
||||
}
|
||||
created, err = s.RegisterSubscription(ctx, "series-1", "別の名前", episodes, true)
|
||||
if err != nil || created {
|
||||
t.Fatalf("重複登録: %t %v", created, err)
|
||||
}
|
||||
if err := pool.QueryRow(ctx, `SELECT count(*) FROM download_jobs WHERE series_id='series-1'`).Scan(&jobs); err != nil || jobs != 0 {
|
||||
t.Fatalf("重複登録が投入した: %d %v", jobs, err)
|
||||
}
|
||||
bad := seriesEpisode("bad", "series-bad")
|
||||
bad.DurationSeconds = -1
|
||||
created, err = s.RegisterSubscription(ctx, "series-bad", "番組", []Episode{seriesEpisode("good", "series-bad"), bad}, true)
|
||||
if err == nil || created {
|
||||
t.Fatalf("壊れた回の登録が成功: %t %v", created, err)
|
||||
}
|
||||
for _, table := range []string{"subscriptions", "known_episodes", "download_jobs"} {
|
||||
var count int
|
||||
if err := pool.QueryRow(ctx, `SELECT count(*) FROM `+table+` WHERE series_id='series-bad'`).Scan(&count); err != nil || count != 0 {
|
||||
t.Errorf("取引の巻き戻し %s: %d %v", table, count, err)
|
||||
}
|
||||
}
|
||||
deleted, err := s.DeleteSubscription(ctx, "series-1")
|
||||
if err != nil || !deleted {
|
||||
t.Fatalf("解除: %t %v", deleted, err)
|
||||
}
|
||||
if err := pool.QueryRow(ctx, `SELECT count(*) FROM known_episodes WHERE series_id='series-1'`).Scan(&jobs); err != nil || jobs != 0 {
|
||||
t.Fatalf("解除後に既知が残った: %d %v", jobs, err)
|
||||
}
|
||||
created, err = s.RegisterSubscription(ctx, "series-1", "番組", episodes, true)
|
||||
if err != nil || !created {
|
||||
t.Fatalf("解除後の再登録: %t %v", created, err)
|
||||
}
|
||||
if err := pool.QueryRow(ctx, `SELECT count(*) FROM download_jobs WHERE series_id='series-1'`).Scan(&jobs); err != nil || jobs != 2 {
|
||||
t.Fatalf("再登録で既存回を選び直せない: %d %v", jobs, err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRegisterAndQueueUnseenEpisodes(t *testing.T) {
|
||||
pool := testPool(t, true)
|
||||
s := New(pool)
|
||||
ctx := context.Background()
|
||||
manual, _, err := s.Enqueue(ctx, seriesEpisode("manual", "series-2"), "manual")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
created, err := s.RegisterSubscription(ctx, "series-2", "番組", []Episode{seriesEpisode("manual", "series-2"), seriesEpisode("past", "series-2")}, true)
|
||||
if err != nil || !created {
|
||||
t.Fatalf("公開中の回を投入: %t %v", created, err)
|
||||
}
|
||||
manualAfter, err := s.GetJob(ctx, manual.ID)
|
||||
if err != nil || manualAfter.Source != "manual" {
|
||||
t.Fatalf("既存の手動ジョブを変更: %+v %v", manualAfter, err)
|
||||
}
|
||||
var subscriptionJobs int
|
||||
if err := pool.QueryRow(ctx, `SELECT count(*) FROM download_jobs WHERE series_id='series-2' AND source='subscription' AND status='queued'`).Scan(&subscriptionJobs); err != nil || subscriptionJobs != 1 {
|
||||
t.Fatalf("登録時に作ったジョブ: %d %v", subscriptionJobs, err)
|
||||
}
|
||||
subs, err := s.ListSubscriptions(ctx)
|
||||
if err != nil || len(subs) != 1 || subs[0].JobCount != 2 || !subs[0].Enabled || subs[0].LastCheckedAt == nil {
|
||||
t.Fatalf("登録一覧と関連ジョブ数: %+v %v", subs, err)
|
||||
}
|
||||
unknown, err := s.UnknownEpisodeIDs(ctx, "series-2", []string{"past", "new", "manual", "new"})
|
||||
if err != nil || len(unknown) != 1 || unknown[0] != "new" {
|
||||
t.Fatalf("未確認の抽出: %v %v", unknown, err)
|
||||
}
|
||||
if err := s.SetSubscriptionEnabled(ctx, "series-2", false); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := s.QueueUnseenEpisodes(ctx, "series-2", []Episode{seriesEpisode("new", "series-2")}); !errors.Is(err, ErrSubscriptionPaused) {
|
||||
t.Fatalf("一時停止中の投入: %v", err)
|
||||
}
|
||||
unknown, err = s.UnknownEpisodeIDs(ctx, "series-2", []string{"new"})
|
||||
if err != nil || len(unknown) != 1 {
|
||||
t.Fatalf("一時停止中に既知となった: %v %v", unknown, err)
|
||||
}
|
||||
if err := s.SetSubscriptionEnabled(ctx, "series-2", true); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
invalid := seriesEpisode("invalid", "series-2")
|
||||
invalid.DurationSeconds = -1
|
||||
if _, err := s.QueueUnseenEpisodes(ctx, "series-2", []Episode{seriesEpisode("newer", "series-2"), invalid}); err == nil {
|
||||
t.Fatal("新規回の投入が途中失敗でも確定した")
|
||||
}
|
||||
unknown, err = s.UnknownEpisodeIDs(ctx, "series-2", []string{"newer"})
|
||||
if err != nil || len(unknown) != 1 {
|
||||
t.Fatalf("新規回の投入失敗が巻き戻らない: %v %v", unknown, err)
|
||||
}
|
||||
count, err := s.QueueUnseenEpisodes(ctx, "series-2", []Episode{seriesEpisode("past", "series-2"), seriesEpisode("new", "series-2")})
|
||||
if err != nil || count != 1 {
|
||||
t.Fatalf("再開後の新しい回: %d %v", count, err)
|
||||
}
|
||||
count, err = s.QueueUnseenEpisodes(ctx, "series-2", []Episode{seriesEpisode("new", "series-2")})
|
||||
if err != nil || count != 0 {
|
||||
t.Fatalf("重複した新しい回: %d %v", count, err)
|
||||
}
|
||||
var source string
|
||||
if err := pool.QueryRow(ctx, `SELECT source FROM download_jobs WHERE episode_id='new'`).Scan(&source); err != nil || source != "subscription" {
|
||||
t.Fatalf("新しい回の登録元: %q %v", source, err)
|
||||
}
|
||||
statuses, err := s.SubscribedSeries(ctx, []string{"series-2", "missing"})
|
||||
if err != nil || !statuses["series-2"] || statuses["missing"] {
|
||||
t.Fatalf("登録済みシリーズの一括取得: %v %v", statuses, err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestQueueUnseenEpisodesConcurrent(t *testing.T) {
|
||||
pool := testPool(t, true)
|
||||
s := New(pool)
|
||||
ctx := context.Background()
|
||||
created, err := s.RegisterSubscription(ctx, "series-concurrent", "番組", nil, false)
|
||||
if err != nil || !created {
|
||||
t.Fatalf("登録: %t %v", created, err)
|
||||
}
|
||||
episodes := []Episode{
|
||||
seriesEpisode("concurrent-1", "series-concurrent"),
|
||||
seriesEpisode("concurrent-2", "series-concurrent"),
|
||||
}
|
||||
const workers = 8
|
||||
counts := make(chan int, workers)
|
||||
errs := make(chan error, workers)
|
||||
var wg sync.WaitGroup
|
||||
for range workers {
|
||||
wg.Add(1)
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
count, err := s.QueueUnseenEpisodes(ctx, "series-concurrent", episodes)
|
||||
counts <- count
|
||||
errs <- err
|
||||
}()
|
||||
}
|
||||
wg.Wait()
|
||||
close(counts)
|
||||
close(errs)
|
||||
for err := range errs {
|
||||
if err != nil {
|
||||
t.Errorf("同時投入: %v", err)
|
||||
}
|
||||
}
|
||||
total := 0
|
||||
positive := 0
|
||||
for count := range counts {
|
||||
total += count
|
||||
if count > 0 {
|
||||
positive++
|
||||
}
|
||||
}
|
||||
if total != 2 || positive != 1 {
|
||||
t.Fatalf("同時投入の記録件数: total=%d positive workers=%d", total, positive)
|
||||
}
|
||||
for table := range map[string]struct{}{"known_episodes": {}, "download_jobs": {}} {
|
||||
var count int
|
||||
if err := pool.QueryRow(ctx, `SELECT count(*) FROM `+table+` WHERE series_id='series-concurrent'`).Scan(&count); err != nil || count != 2 {
|
||||
t.Errorf("%s の重複防止: %d %v", table, count, err)
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user