245 lines
7.9 KiB
Go
245 lines
7.9 KiB
Go
package store
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
"time"
|
|
|
|
"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
|
|
}
|
|
|
|
// ListDueSubscriptions は有効で、cutoff 以前に確認された（または未確認の）番組を返す。
|
|
func (s *Store) ListDueSubscriptions(ctx context.Context, cutoff time.Time) ([]Subscription, error) {
|
|
rows, err := s.pool.Query(ctx, `SELECT series_id, series_title, enabled, created_at, last_checked_at,
|
|
last_check_error_at, last_check_error, 0::bigint
|
|
FROM subscriptions
|
|
WHERE enabled AND (last_checked_at IS NULL OR last_checked_at <= $1)
|
|
ORDER BY last_checked_at ASC NULLS FIRST, created_at, series_id`, cutoff)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
var result []Subscription
|
|
for rows.Next() {
|
|
var sub Subscription
|
|
if err := rows.Scan(&sub.SeriesID, &sub.SeriesTitle, &sub.Enabled, &sub.CreatedAt,
|
|
&sub.LastCheckedAt, &sub.LastCheckErrorAt, &sub.LastCheckError, &sub.JobCount); err != nil {
|
|
return nil, err
|
|
}
|
|
result = append(result, sub)
|
|
}
|
|
return result, rows.Err()
|
|
}
|
|
|
|
// CompleteSubscriptionCheck は全公開回を既知にし、しきい値を超える未確認回だけを投入する。
|
|
// 最終確認日時の更新と前回エラーの消去も同じ取引で行う。
|
|
func (s *Store) CompleteSubscriptionCheck(ctx context.Context, seriesID string, episodes []Episode, previewMaxSeconds int, checkedAt time.Time) (int, error) {
|
|
if previewMaxSeconds < 0 {
|
|
return 0, errors.New("preview threshold must not be negative")
|
|
}
|
|
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
|
|
}
|
|
queued := 0
|
|
for _, ep := range episodes {
|
|
if ep.ID == "" || ep.SeriesID != seriesID || ep.DurationSeconds < 0 {
|
|
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
|
|
}
|
|
if ep.DurationSeconds <= previewMaxSeconds {
|
|
continue
|
|
}
|
|
tag, 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
|
|
}
|
|
queued += int(tag.RowsAffected())
|
|
}
|
|
if _, err := tx.Exec(ctx, `UPDATE subscriptions SET last_checked_at=$2,
|
|
last_check_error_at=NULL, last_check_error=NULL WHERE series_id=$1`, seriesID, checkedAt); err != nil {
|
|
return 0, err
|
|
}
|
|
if err := tx.Commit(ctx); err != nil {
|
|
return 0, err
|
|
}
|
|
return queued, nil
|
|
}
|
|
|
|
// RecordSubscriptionCheckFailure は取得失敗を記録し、最終確認日時は変更しない。
|
|
func (s *Store) RecordSubscriptionCheckFailure(ctx context.Context, seriesID string, failedAt time.Time, message string) error {
|
|
tag, err := s.pool.Exec(ctx, `UPDATE subscriptions SET last_check_error_at=$2, last_check_error=$3
|
|
WHERE series_id=$1 AND enabled`, seriesID, failedAt, message)
|
|
return transitionResult(tag.RowsAffected(), err)
|
|
}
|
|
|
|
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()
|
|
}
|