thamanyah/discovery/internal/db/repositories/videos.go

package repositories

import (
	"context"
	"database/sql"
	"encoding/base64"
	"encoding/json"
	"errors"
	"strings"
	"thamanyah/discovery/internal/models"
	"time"
	"unicode"

	"github.com/lib/pq"
)

type VideoRepository struct {
	SQLDB *sql.DB
}

var VideoRepo VideoRepository

// SaveVideo records an announced video, replacing what the catalogue already
// held for that id.
//
// The upsert is what makes the consumer safe to retry: the catalogue topic
// delivers at-least-once, so the same announcement can arrive twice, and a
// plain INSERT would fail the second time on the primary key. Re-announcing a
// video is also how it is corrected — a later announcement wins, which is why
// every column is overwritten rather than merged.
func (svc VideoRepository) SaveVideo(ctx context.Context, v models.Video) error {
	_, err := svc.SQLDB.ExecContext(ctx, `
		INSERT INTO videos (id, title, playback_url, categories)
		VALUES ($1, $2, $3, $4)
		ON CONFLICT (id) DO UPDATE
		SET title = EXCLUDED.title,
		    playback_url = EXCLUDED.playback_url,
		    categories = EXCLUDED.categories,
		    updated_at = now()
	`, v.ID, v.Title, v.PlaybackURL, pq.Array(v.Categories))
	return err
}

// GetVideoByID reads the catalogue's copy of one video. It returns
// ErrVideoNotFound when the catalogue has not heard about it — which, for a
// video cms has only just made ready, is a matter of timing rather than a
// mistake.
func (svc VideoRepository) GetVideoByID(ctx context.Context, id string) (models.Video, error) {
	var v models.Video

	err := svc.SQLDB.QueryRowContext(ctx, `
		SELECT id, title, playback_url, categories, created_at, updated_at
		FROM videos
		WHERE id = $1
	`, id).Scan(&v.ID, &v.Title, &v.PlaybackURL, pq.Array(&v.Categories), &v.CreatedAt, &v.UpdatedAt)
	if errors.Is(err, sql.ErrNoRows) {
		return models.Video{}, ErrVideoNotFound
	}
	if err != nil {
		return models.Video{}, err
	}

	return v, nil
}

// searchConfig is the text search configuration the search_vector column is
// generated with. Every query has to name the same one: a tsquery built under
// a different configuration stems its terms differently and matches nothing.
const searchConfig = "arabic"

// tsQueryFor turns what a reader typed into a tsquery.
//
// The words are ANDed, and the last one is given a :* prefix match — someone
// typing into a search box is usually part-way through their last word, so
// "desert fal" should still find "Desert Falcons". Earlier words are matched
// whole, because they have been finished.
//
// The input is split on everything that is not a letter or a digit, which
// drops every character tsquery gives a meaning to (&, |, !, :, *, parens) and
// leaves nothing that could change the shape of the query. That is what makes
// it safe to interpolate the result into to_tsquery: the alternative,
// websearch_to_tsquery, parses user syntax but cannot express a prefix match.
//
// It returns "" when there is nothing to search for, which callers read as
// "no title filter" rather than "match nothing".
func tsQueryFor(term string) string {
	words := strings.FieldsFunc(term, func(r rune) bool {
		return !unicode.IsLetter(r) && !unicode.IsDigit(r)
	})
	if len(words) == 0 {
		return ""
	}

	for i, word := range words {
		words[i] = strings.ToLower(word)
	}
	return strings.Join(words, " & ") + ":*"
}

// searchCursor is where a page of results stopped: the sort key of its last
// row. Paging resumes by asking for the rows that sort after it.
//
// It carries the rank as well as the row's identity because the results are
// ordered by relevance first, and "the row after this one" is only a
// well-defined place if every term of the ordering is pinned.
type searchCursor struct {
	Rank      float64   `json:"r"`
	CreatedAt time.Time `json:"t"`
	ID        string    `json:"i"`
}

// encode renders a cursor as an opaque string. Opaque on purpose: it is a
// place in a result set, not a number a client should do arithmetic on, and
// making it look like one invites callers to guess a page.
func (c searchCursor) encode() (string, error) {
	raw, err := json.Marshal(c)
	if err != nil {
		return "", err
	}
	return base64.RawURLEncoding.EncodeToString(raw), nil
}

func decodeCursor(encoded string) (searchCursor, error) {
	raw, err := base64.RawURLEncoding.DecodeString(encoded)
	if err != nil {
		return searchCursor{}, ErrInvalidCursor
	}

	var c searchCursor
	if err := json.Unmarshal(raw, &c); err != nil {
		return searchCursor{}, ErrInvalidCursor
	}
	if strings.TrimSpace(c.ID) == "" {
		return searchCursor{}, ErrInvalidCursor
	}

	return c, nil
}

// SearchVideos finds catalogued videos whose title matches the given term and
// which are filed under any of the given categories, most relevant first.
//
// It returns the page and the cursor that reaches the page after it; that
// cursor is empty on the last page, which is how a caller knows to stop.
//
// The title predicate is skipped entirely when the term yields no words, and
// the category predicate when no categories are named, so a search with
// neither browses the whole catalogue newest-first.
//
// Paging is by key rather than by offset. OFFSET makes the database walk and
// discard every row it skips, so deep pages get steadily more expensive, and
// it addresses a page by its position — which moves whenever a video is
// announced, repeating or skipping rows for a reader mid-browse. Comparing
// against the last row's sort key has neither problem: the work is the same at
// any depth, and the page after a given row is that same page however much has
// been added since.
func (svc VideoRepository) SearchVideos(ctx context.Context, term string, categories []string, limit int, cursor string) ([]models.Video, string, error) {
	query := tsQueryFor(term)

	// Never nil: a nil slice reaches Postgres as NULL, and cardinality(NULL)
	// is NULL rather than 0, so the "no category filter" branch would not fire
	// and every search would come back empty.
	if categories == nil {
		categories = []string{}
	}

	// A zero cursor is passed as NULL, which the query reads as "start at the
	// beginning" rather than as a row to seek past.
	var after *searchCursor
	if strings.TrimSpace(cursor) != "" {
		decoded, err := decodeCursor(cursor)
		if err != nil {
			return nil, "", err
		}
		after = &decoded
	}

	var (
		afterRank *float64
		afterTime *time.Time
		afterID   *string
	)
	if after != nil {
		afterRank, afterTime, afterID = &after.Rank, &after.CreatedAt, &after.ID
	}

	// One row more than asked for: if it comes back, there is another page and
	// the extra row is discarded. That is what lets the last page say so,
	// rather than handing out a cursor to an empty page.
	//
	// The rank is computed in the subquery and referenced by name in both the
	// cursor comparison and the ORDER BY, so the two cannot drift apart — if
	// they did, "the row after this one" would mean something different from
	// the order the rows are actually in, and pages would overlap.
	rows, err := svc.SQLDB.QueryContext(ctx, `
		SELECT id, title, playback_url, categories, created_at, rank
		FROM (
			SELECT id, title, playback_url, categories, created_at,
			       CASE WHEN $1 = '' THEN 0::float8
			            ELSE ts_rank(search_vector, to_tsquery('`+searchConfig+`', $1))::float8
			       END AS rank
			FROM videos
			WHERE ($1 = '' OR search_vector @@ to_tsquery('`+searchConfig+`', $1))
			  AND (cardinality($2::text[]) = 0 OR categories && $2)
		) ranked
		WHERE $4::float8 IS NULL
		   OR (rank, created_at, id) < ($4::float8, $5::timestamptz, $6::uuid)
		ORDER BY rank DESC, created_at DESC, id DESC
		LIMIT $3
	`, query, pq.Array(categories), limit+1, afterRank, afterTime, afterID)
	if err != nil {
		return nil, "", err
	}
	defer rows.Close()

	// Never nil: a search that matches nothing is an empty list on the wire,
	// not null.
	videos := []models.Video{}
	ranks := []float64{}
	for rows.Next() {
		var (
			v    models.Video
			rank float64
		)
		if err := rows.Scan(&v.ID, &v.Title, &v.PlaybackURL, pq.Array(&v.Categories), &v.CreatedAt, &rank); err != nil {
			return nil, "", err
		}
		videos = append(videos, v)
		ranks = append(ranks, rank)
	}
	if err := rows.Err(); err != nil {
		return nil, "", err
	}

	if len(videos) <= limit {
		return videos, "", nil
	}

	videos, ranks = videos[:limit], ranks[:limit]
	last := videos[len(videos)-1]

	next, err := searchCursor{
		Rank:      ranks[len(ranks)-1],
		CreatedAt: last.CreatedAt,
		ID:        last.ID,
	}.encode()
	if err != nil {
		return nil, "", err
	}

	return videos, next, nil
}