melosh/mal
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: ensure worker queue advances on error

Browse Source
This commit is contained in:
mkelvers
2026-04-08 15:00:19 +02:00
parent 5f4ed88d1a
commit a5b72c599c
4 changed files with 57 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
internal/database/querier.go
View File

@@ -23,6 +23,7 @@ type Querier interface {
GetUserWatchList(ctx context.Context, userID string) ([]GetUserWatchListRow, error) GetUserWatchList(ctx context.Context, userID string) ([]GetUserWatchListRow, error)
GetWatchListEntry(ctx context.Context, arg GetWatchListEntryParams) (WatchListEntry, error) GetWatchListEntry(ctx context.Context, arg GetWatchListEntryParams) (WatchListEntry, error)
GetWatchingAnime(ctx context.Context, userID string) ([]GetWatchingAnimeRow, error) GetWatchingAnime(ctx context.Context, userID string) ([]GetWatchingAnimeRow, error)
MarkRelationsSynced(ctx context.Context, id int64) error
UpdateAnimeStatus(ctx context.Context, arg UpdateAnimeStatusParams) error UpdateAnimeStatus(ctx context.Context, arg UpdateAnimeStatusParams) error
UpsertAnime(ctx context.Context, arg UpsertAnimeParams) (Anime, error) UpsertAnime(ctx context.Context, arg UpsertAnimeParams) (Anime, error)
UpsertAnimeRelation(ctx context.Context, arg UpsertAnimeRelationParams) error UpsertAnimeRelation(ctx context.Context, arg UpsertAnimeRelationParams) error

5
internal/database/queries.sql
View File

@@ -86,7 +86,10 @@ ON CONFLICT (anime_id, related_anime_id) DO UPDATE SET
relation_type = excluded.relation_type; relation_type = excluded.relation_type;
-- name: UpdateAnimeStatus :exec -- name: UpdateAnimeStatus :exec
UPDATE anime SET status = ?, relations_synced_at = CURRENT_TIMESTAMP WHERE id = ?; UPDATE anime SET status = ? WHERE id = ?;
-- name: MarkRelationsSynced :exec
UPDATE anime SET relations_synced_at = CURRENT_TIMESTAMP WHERE id = ?;
-- name: GetAnimeNeedingRelationSync :many -- name: GetAnimeNeedingRelationSync :many
WITH RECURSIVE sequel_chain AS ( WITH RECURSIVE sequel_chain AS (

11
internal/database/queries.sql.go
View File

@@ -459,8 +459,17 @@ func (q *Queries) GetWatchingAnime(ctx context.Context, userID string) ([]GetWat
return items, nil return items, nil
} }
const markRelationsSynced = `-- name: MarkRelationsSynced :exec
UPDATE anime SET relations_synced_at = CURRENT_TIMESTAMP WHERE id = ?
`
func (q *Queries) MarkRelationsSynced(ctx context.Context, id int64) error {
_, err := q.db.ExecContext(ctx, markRelationsSynced, id)
return err
}
const updateAnimeStatus = `-- name: UpdateAnimeStatus :exec const updateAnimeStatus = `-- name: UpdateAnimeStatus :exec
UPDATE anime SET status = ?, relations_synced_at = CURRENT_TIMESTAMP WHERE id = ? UPDATE anime SET status = ? WHERE id = ?
` `
type UpdateAnimeStatusParams struct { type UpdateAnimeStatusParams struct {

76
internal/worker/relations.go
View File

@@ -51,47 +51,55 @@ func (w *Worker) syncRelations(ctx context.Context) {
for _, a := range animes { for _, a := range animes {
log.Printf("worker: syncing relations for anime %d (%s)", a.ID, a.TitleOriginal) log.Printf("worker: syncing relations for anime %d (%s)", a.ID, a.TitleOriginal)
relations, err := w.client.GetRelationsData(int(a.ID)) func() {
if err != nil { // Always mark as synced and sleep so the queue advances even on error.
log.Printf("worker: failed to fetch relations for %d: %v", a.ID, err) defer func() {
continue err := w.db.MarkRelationsSynced(ctx, a.ID)
} if err != nil {
log.Printf("worker: failed to mark relations synced for %d: %v", a.ID, err)
}
time.Sleep(400 * time.Millisecond)
}()
for _, rel := range relations.Data { relations, err := w.client.GetRelationsData(int(a.ID))
for _, entry := range rel.Entry { if err != nil {
if entry.Type == "anime" { log.Printf("worker: failed to fetch relations for %d: %v", a.ID, err)
// We just insert the relation. return
err := w.db.UpsertAnimeRelation(ctx, database.UpsertAnimeRelationParams{ }
AnimeID: a.ID,
RelatedAnimeID: int64(entry.MalID),
RelationType: rel.Relation,
})
if err != nil {
log.Printf("worker: failed to insert relation %d -> %d: %v", a.ID, entry.MalID, err)
}
// If it's a Sequel, we should also make sure the related anime is tracked for _, rel := range relations.Data {
if rel.Relation == "Sequel" { for _, entry := range rel.Entry {
w.ensureAnimeExistsAndStatusUpdated(ctx, entry.MalID) if entry.Type == "anime" {
// We just insert the relation.
err := w.db.UpsertAnimeRelation(ctx, database.UpsertAnimeRelationParams{
AnimeID: a.ID,
RelatedAnimeID: int64(entry.MalID),
RelationType: rel.Relation,
})
if err != nil {
log.Printf("worker: failed to insert relation %d -> %d: %v", a.ID, entry.MalID, err)
}
// If it's a Sequel, we should also make sure the related anime is tracked
if rel.Relation == "Sequel" {
w.ensureAnimeExistsAndStatusUpdated(ctx, entry.MalID)
}
} }
} }
} }
}
// Also update the status of the anime itself so we know if it's Not yet aired, etc. // Also update the status of the anime itself so we know if it's Not yet aired, etc.
animeDetails, err := w.client.GetAnimeByID(int(a.ID)) animeDetails, err := w.client.GetAnimeByID(int(a.ID))
if err == nil { if err == nil {
err = w.db.UpdateAnimeStatus(ctx, database.UpdateAnimeStatusParams{ err = w.db.UpdateAnimeStatus(ctx, database.UpdateAnimeStatusParams{
Status: sql.NullString{String: animeDetails.Status, Valid: true}, Status: sql.NullString{String: animeDetails.Status, Valid: true},
ID: a.ID, ID: a.ID,
}) })
if err != nil { if err != nil {
log.Printf("worker: failed to update status for %d: %v", a.ID, err) log.Printf("worker: failed to update status for %d: %v", a.ID, err)
}
} }
} }()
// Sleep briefly to respect Jikan's 3 req/sec rate limit
time.Sleep(400 * time.Millisecond)
} }
} }