minstrel/internal/requests/reconciler.go

package requests

import (
	"context"
	"errors"
	"log/slog"
	"time"

	"github.com/jackc/pgx/v5"
	"github.com/jackc/pgx/v5/pgtype"
	"github.com/jackc/pgx/v5/pgxpool"

	"git.fabledsword.com/bvandeusen/minstrel/internal/db/dbq"
	"git.fabledsword.com/bvandeusen/minstrel/internal/eventbus"
	"git.fabledsword.com/bvandeusen/minstrel/internal/notifications"
	"git.fabledsword.com/bvandeusen/minstrel/internal/wanted"
)

// Reconciler is a background worker that completes approved requests once
// the library holds what they asked for: an artist request when the artist
// is in the library, an album or track request when the want it recorded is
// held (internal/wanted settles wants). Errors per row are logged at WARN and
// do not abort the tick.
type Reconciler struct {
	pool     *pgxpool.Pool
	logger   *slog.Logger
	bus      *eventbus.Bus
	notifier *notifications.Notifier // nil: no inbox notifications
	tick     time.Duration
	batch    int32
}

// NewReconciler constructs a Reconciler with production defaults: 5-minute
// tick, batch size 50. Pass nil for bus when no SSE publishing is desired.
func NewReconciler(pool *pgxpool.Pool, logger *slog.Logger, bus *eventbus.Bus) *Reconciler {
	return &Reconciler{pool: pool, logger: logger, bus: bus, tick: 5 * time.Minute, batch: 50}
}

// SetNotifier makes completions land in the requester's notifications inbox.
// Without one, completions are only broadcast on the bus.
func (r *Reconciler) SetNotifier(n *notifications.Notifier) { r.notifier = n }

// publishCompleted tells the requester their request arrived: a
// request.status_changed event for open screens, and a notification that
// outlives the connection.
func (r *Reconciler) publishCompleted(ctx context.Context, row dbq.Request) {
	p := notifications.Payload{
		RequestID:   formatUUIDForBus(row.ID),
		RequestKind: string(row.Kind),
		Name:        DisplayName(row),
		Artist:      row.ArtistName,
		Title:       requestedTitle(row),
	}
	if row.MatchedAlbumID.Valid {
		p.AlbumID = formatUUIDForBus(row.MatchedAlbumID)
	} else if row.MatchedArtistID.Valid {
		p.ArtistID = formatUUIDForBus(row.MatchedArtistID)
	}
	r.notifier.NotifyLogged(ctx, notifications.KindRequestCompleted, notifications.ToUser(row.UserID), p.Map())

	if r.bus == nil {
		return
	}
	r.bus.Publish(eventbus.Event{
		Kind:   "request.status_changed",
		UserID: formatUUIDForBus(row.UserID),
		Data: map[string]any{
			"request_id": formatUUIDForBus(row.ID),
			"status":     "completed",
			"kind":       string(row.Kind),
		},
	})
}

// formatUUIDForBus renders a pgtype.UUID as the canonical 8-4-4-4-12 hex
// string. Mirrors helpers in internal/api/convert.go and other packages;
// inlined here to avoid a back-edge dependency from requests onto
// the api layer.
func formatUUIDForBus(u pgtype.UUID) string {
	if !u.Valid {
		return ""
	}
	const hex = "0123456789abcdef"
	out := make([]byte, 36)
	pos := 0
	for i, b := range u.Bytes {
		if i == 4 || i == 6 || i == 8 || i == 10 {
			out[pos] = '-'
			pos++
		}
		out[pos] = hex[b>>4]
		out[pos+1] = hex[b&0xf]
		pos += 2
	}
	return string(out)
}

// Run blocks until ctx is cancelled, ticking every r.tick. Errors from
// tickOnce are logged at WARN and never propagated.
func (r *Reconciler) Run(ctx context.Context) {
	t := time.NewTicker(r.tick)
	defer t.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		case <-t.C:
			if err := r.tickOnce(ctx); err != nil {
				r.logger.Warn("requests: reconciler tick failed", "err", err)
			}
		}
	}
}

// tickOnce loads approved requests (oldest first, up to r.batch) and
// completes those the library now holds.
func (r *Reconciler) tickOnce(ctx context.Context) error {
	q := dbq.New(r.pool)
	rows, err := q.ListApprovedRequestsForReconcile(ctx, r.batch)
	if err != nil {
		return err
	}
	for _, row := range rows {
		if err := r.reconcileRow(ctx, q, row); err != nil {
			r.logger.Warn("requests: reconciler row failed",
				"request_id", row.ID, "kind", row.Kind, "err", err)
		}
	}
	return nil
}

// reconcileRow completes one request if the library holds it. A request
// not yet held is not an error.
func (r *Reconciler) reconcileRow(ctx context.Context, q *dbq.Queries, row dbq.Request) error {
	if row.Kind == dbq.RequestKindArtist {
		return r.reconcileArtist(ctx, q, row)
	}
	return r.reconcileAlbum(ctx, q, row)
}

func (r *Reconciler) reconcileArtist(ctx context.Context, q *dbq.Queries, row dbq.Request) error {
	var artistID pgtype.UUID
	err := r.pool.QueryRow(ctx, "SELECT id FROM artists WHERE mbid = $1", row.ArtistMbid).Scan(&artistID)
	if errors.Is(err, pgx.ErrNoRows) {
		return nil // not in library yet
	}
	if err != nil {
		return err
	}
	completed, err := q.CompleteRequest(ctx, dbq.CompleteRequestParams{ID: row.ID, MatchedArtistID: artistID})
	if err != nil {
		return err
	}
	r.publishCompleted(ctx, completed)
	return nil
}

// reconcileAlbum completes an album or track request once its want is held,
// matched to the album that answered it. A track request is matched to the
// track it named when that recording arrived, else the newest arrival.
func (r *Reconciler) reconcileAlbum(ctx context.Context, q *dbq.Queries, row dbq.Request) error {
	if row.AlbumMbid == nil || *row.AlbumMbid == "" {
		return nil
	}
	w, err := q.GetWantByReleaseGroup(ctx, *row.AlbumMbid)
	if errors.Is(err, pgx.ErrNoRows) {
		// Removed from the wanted list while the request waited. Wanting it
		// again keeps the approval meaningful; the admin who removed it can
		// reject a request instead.
		_, err = wanted.WantWith(ctx, q, wantFor(row))
		return err
	}
	if err != nil {
		return err
	}
	if !w.FulfilledAt.Valid || !w.AlbumID.Valid {
		return nil
	}
	params := dbq.CompleteRequestParams{ID: row.ID, MatchedAlbumID: w.AlbumID}
	if row.Kind == dbq.RequestKindTrack {
		var trackID pgtype.UUID
		err := r.pool.QueryRow(ctx, `
			SELECT id FROM tracks
			 WHERE album_id = $1 AND missing_since IS NULL
			 ORDER BY (mbid IS NOT DISTINCT FROM $2) DESC, added_at DESC, id
			 LIMIT 1`, w.AlbumID, row.TrackMbid).Scan(&trackID)
		if err != nil && !errors.Is(err, pgx.ErrNoRows) {
			return err
		}
		params.MatchedTrackID = trackID
	}
	completed, err := q.CompleteRequest(ctx, params)
	if err != nil {
		return err
	}
	r.publishCompleted(ctx, completed)
	return nil
}