iu
96268ee16c
Move physical file facts out of the photos table into an objects ledger:
- objects / photo_objects / variant_jobs tables (migration 000008)
- drop storage_key/thumb_key/preview_key/size/width/height/mime_type from
photos (migration 000009); they are derived via joins to the original object
- album-free storage keys: objects/{id}/{role}.{ext}
- explicit GC via pending/ready/deleting states; objects deleted only when no
photo references them
- background worker drives variant generation (durable variant_jobs) and GC
- copy/move are metadata-only; storage objects are shared between photos
- remove unused storage.DeletePrefix
Non-destructive to the external JSON API: size/dimensions/mime and URLs are
still returned, just sourced from the object layer.
OpenSpec change: media-object-layer
498 lines
15 KiB
Go
498 lines
15 KiB
Go
package photos
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/jackc/pgx/v5"
|
|
"github.com/jackc/pgx/v5/pgconn"
|
|
"github.com/jackc/pgx/v5/pgxpool"
|
|
|
|
"shared-album/internal/auth"
|
|
"shared-album/internal/models"
|
|
)
|
|
|
|
var ErrNotFound = errors.New("photo not found")
|
|
|
|
type Store struct {
|
|
pool *pgxpool.Pool
|
|
}
|
|
|
|
func NewStore(pool *pgxpool.Pool) *Store { return &Store{pool: pool} }
|
|
|
|
// photoColumns selects photo rows together with the keys and physical facts of
|
|
// their bound objects (original, thumb, preview) and the original's size,
|
|
// dimensions and mime type.
|
|
const photoColumns = `p.id, COALESCE(p.album_id, ''), p.uploader_id,
|
|
COALESCE(o_o.storage_key, ''),
|
|
COALESCE(o_t.storage_key, ''),
|
|
COALESCE(o_p.storage_key, ''),
|
|
p.original_filename, COALESCE(o_o.size, 0), COALESCE(o_o.width, 0), COALESCE(o_o.height, 0), COALESCE(o_o.mime_type, ''),
|
|
p.taken_at, p.latitude, p.longitude, COALESCE(p.caption, ''), p.created_at,
|
|
u.id, u.name, COALESCE(u.avatar_url, ''),
|
|
(SELECT count(*) FROM likes l WHERE l.photo_id = p.id),
|
|
(SELECT count(*) FROM comments c WHERE c.photo_id = p.id),
|
|
EXISTS(SELECT 1 FROM likes l WHERE l.photo_id = p.id AND l.user_id = $2),
|
|
COALESCE(p.exif, '{}'::jsonb)::text, p.deleted_at,
|
|
COALESCE((SELECT json_agg(json_build_object('id', t.id, 'name', t.name) ORDER BY lower(t.name))
|
|
FROM photo_tags pt JOIN tags t ON t.id = pt.tag_id
|
|
WHERE pt.photo_id = p.id AND t.user_id = $2), '[]')::text`
|
|
|
|
const photoJoins = `
|
|
LEFT JOIN photo_objects po_o ON po_o.photo_id = p.id AND po_o.role = 'original'
|
|
LEFT JOIN objects o_o ON o_o.id = po_o.object_id
|
|
LEFT JOIN photo_objects po_t ON po_t.photo_id = p.id AND po_t.role = 'thumb'
|
|
LEFT JOIN objects o_t ON o_t.id = po_t.object_id
|
|
LEFT JOIN photo_objects po_p ON po_p.photo_id = p.id AND po_p.role = 'preview'
|
|
LEFT JOIN objects o_p ON o_p.id = po_p.object_id`
|
|
|
|
func scanPhoto(row pgx.Row) (*models.Photo, error) {
|
|
var p models.Photo
|
|
var exifText, tagsText string
|
|
err := row.Scan(&p.ID, &p.AlbumID, &p.UploaderID, &p.StorageKey,
|
|
&p.ThumbKey, &p.PreviewKey,
|
|
&p.OriginalFilename, &p.Size, &p.Width, &p.Height, &p.MimeType,
|
|
&p.TakenAt, &p.Latitude, &p.Longitude, &p.Caption, &p.CreatedAt,
|
|
&p.Uploader.ID, &p.Uploader.Name, &p.Uploader.AvatarURL,
|
|
&p.LikesCount, &p.CommentsCount, &p.LikedByMe, &exifText, &p.DeletedAt, &tagsText)
|
|
if err != nil {
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|
return nil, ErrNotFound
|
|
}
|
|
return nil, err
|
|
}
|
|
setExif(&p, exifText)
|
|
setTags(&p, tagsText)
|
|
return &p, nil
|
|
}
|
|
|
|
func setTags(p *models.Photo, raw string) {
|
|
if raw == "" || raw == "null" || raw == "[]" {
|
|
return
|
|
}
|
|
var tags []models.Tag
|
|
if err := json.Unmarshal([]byte(raw), &tags); err == nil && len(tags) > 0 {
|
|
p.Tags = tags
|
|
}
|
|
}
|
|
|
|
func setExif(p *models.Photo, raw string) {
|
|
if raw == "" || raw == "null" || raw == "{}" {
|
|
return
|
|
}
|
|
var m map[string]any
|
|
if err := json.Unmarshal([]byte(raw), &m); err == nil && len(m) > 0 {
|
|
p.EXIF = m
|
|
}
|
|
}
|
|
|
|
// CreateWithOriginal records a new photo together with its pending original
|
|
// object, the original binding and a variant job — all in one transaction. The
|
|
// caller stores the bytes and then activates the object. The metadata layer is
|
|
// the ledger of record; the storage layer follows.
|
|
func (s *Store) CreateWithOriginal(ctx context.Context, p *models.Photo, obj *Object) error {
|
|
exifJSON, err := json.Marshal(p.EXIF)
|
|
if err != nil || len(exifJSON) == 0 {
|
|
exifJSON = []byte("{}")
|
|
}
|
|
tx, err := s.pool.Begin(ctx)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer func() { _ = tx.Rollback(ctx) }()
|
|
|
|
if _, err := tx.Exec(ctx,
|
|
`INSERT INTO objects (id, storage_key, mime_type, size, width, height, checksum, state)
|
|
VALUES ($1, $2, $3, $4, $5, $6, $7, $8)`,
|
|
obj.ID, obj.StorageKey, obj.MimeType, obj.Size, obj.Width, obj.Height, obj.Checksum, ObjectPending); err != nil {
|
|
return err
|
|
}
|
|
if err := tx.QueryRow(ctx,
|
|
`INSERT INTO photos (id, album_id, uploader_id, original_filename, taken_at, latitude, longitude, caption, exif)
|
|
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9::jsonb)
|
|
RETURNING created_at`,
|
|
p.ID, p.AlbumID, p.UploaderID, p.OriginalFilename, p.TakenAt, p.Latitude, p.Longitude, p.Caption, string(exifJSON),
|
|
).Scan(&p.CreatedAt); err != nil {
|
|
return err
|
|
}
|
|
if _, err := tx.Exec(ctx,
|
|
`INSERT INTO photo_objects (photo_id, object_id, role) VALUES ($1, $2, $3)`,
|
|
p.ID, obj.ID, RoleOriginal); err != nil {
|
|
return err
|
|
}
|
|
if _, err := tx.Exec(ctx,
|
|
`INSERT INTO variant_jobs (photo_id) VALUES ($1) ON CONFLICT (photo_id) DO NOTHING`, p.ID); err != nil {
|
|
return err
|
|
}
|
|
return tx.Commit(ctx)
|
|
}
|
|
|
|
// DeleteUpload rolls back a failed upload: the photo (and its bindings and job)
|
|
// plus the pending object row are removed.
|
|
func (s *Store) DeleteUpload(ctx context.Context, photoID, objectID string) error {
|
|
tx, err := s.pool.Begin(ctx)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer func() { _ = tx.Rollback(ctx) }()
|
|
if _, err := tx.Exec(ctx, `DELETE FROM photos WHERE id = $1`, photoID); err != nil {
|
|
return err
|
|
}
|
|
if _, err := tx.Exec(ctx, `DELETE FROM objects WHERE id = $1`, objectID); err != nil {
|
|
return err
|
|
}
|
|
return tx.Commit(ctx)
|
|
}
|
|
|
|
func (s *Store) GetByID(ctx context.Context, id, viewerID string) (*models.Photo, error) {
|
|
return scanPhoto(s.pool.QueryRow(ctx,
|
|
`SELECT `+photoColumns+`
|
|
FROM photos p JOIN users u ON u.id = p.uploader_id`+photoJoins+`
|
|
WHERE p.id = $1`, id, viewerID))
|
|
}
|
|
|
|
type ListParams struct {
|
|
AlbumID string
|
|
Viewer string
|
|
Cursor string
|
|
Since *time.Time
|
|
Limit int
|
|
}
|
|
|
|
func (s *Store) List(ctx context.Context, in ListParams) ([]models.Photo, error) {
|
|
if in.Limit <= 0 || in.Limit > 100 {
|
|
in.Limit = 50
|
|
}
|
|
rows, err := s.pool.Query(ctx,
|
|
`SELECT `+photoColumns+`
|
|
FROM photos p JOIN users u ON u.id = p.uploader_id`+photoJoins+`
|
|
WHERE p.album_id = $1
|
|
AND p.deleted_at IS NULL
|
|
AND ($3 = '' OR (p.created_at, p.id) < (SELECT created_at, id FROM photos WHERE id = $3))
|
|
AND ($4::timestamptz IS NULL OR p.created_at > $4)
|
|
ORDER BY p.created_at DESC, p.id DESC
|
|
LIMIT $5`,
|
|
in.AlbumID, in.Viewer, in.Cursor, in.Since, in.Limit)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
|
|
out := []models.Photo{}
|
|
for rows.Next() {
|
|
var p models.Photo
|
|
var exifText, tagsText string
|
|
if err := rows.Scan(&p.ID, &p.AlbumID, &p.UploaderID, &p.StorageKey,
|
|
&p.ThumbKey, &p.PreviewKey,
|
|
&p.OriginalFilename, &p.Size, &p.Width, &p.Height, &p.MimeType,
|
|
&p.TakenAt, &p.Latitude, &p.Longitude, &p.Caption, &p.CreatedAt,
|
|
&p.Uploader.ID, &p.Uploader.Name, &p.Uploader.AvatarURL,
|
|
&p.LikesCount, &p.CommentsCount, &p.LikedByMe, &exifText, &p.DeletedAt, &tagsText); err != nil {
|
|
return nil, err
|
|
}
|
|
setExif(&p, exifText)
|
|
setTags(&p, tagsText)
|
|
out = append(out, p)
|
|
}
|
|
return out, rows.Err()
|
|
}
|
|
|
|
func (s *Store) SetDeleted(ctx context.Context, id string, deleted bool) error {
|
|
var tag pgconn.CommandTag
|
|
var err error
|
|
if deleted {
|
|
tag, err = s.pool.Exec(ctx,
|
|
`UPDATE photos SET deleted_at = now() WHERE id = $1 AND deleted_at IS NULL`, id)
|
|
} else {
|
|
tag, err = s.pool.Exec(ctx,
|
|
`UPDATE photos SET deleted_at = NULL WHERE id = $1`, id)
|
|
}
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if tag.RowsAffected() == 0 {
|
|
return ErrNotFound
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (s *Store) UpdateAlbum(ctx context.Context, id, albumID string) error {
|
|
tag, err := s.pool.Exec(ctx,
|
|
`UPDATE photos SET album_id = $2 WHERE id = $1 AND deleted_at IS NULL`, id, albumID)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if tag.RowsAffected() == 0 {
|
|
return ErrNotFound
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// Clone creates a new photo row in albumID that references the same objects as
|
|
// src. No storage operation is performed: the clone shares the physical files.
|
|
func (s *Store) Clone(ctx context.Context, src *models.Photo, albumID, uploaderID, id string) error {
|
|
exifJSON, err := json.Marshal(src.EXIF)
|
|
if err != nil || len(exifJSON) == 0 {
|
|
exifJSON = []byte("{}")
|
|
}
|
|
tx, err := s.pool.Begin(ctx)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer func() { _ = tx.Rollback(ctx) }()
|
|
|
|
if _, err := tx.Exec(ctx,
|
|
`INSERT INTO photos (id, album_id, uploader_id, original_filename, taken_at, latitude, longitude, caption, exif)
|
|
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9::jsonb)`,
|
|
id, albumID, uploaderID, src.OriginalFilename, src.TakenAt, src.Latitude, src.Longitude, src.Caption, string(exifJSON)); err != nil {
|
|
return err
|
|
}
|
|
if _, err := tx.Exec(ctx,
|
|
`INSERT INTO photo_objects (photo_id, object_id, role)
|
|
SELECT $1, object_id, role FROM photo_objects WHERE photo_id = $2`,
|
|
id, src.ID); err != nil {
|
|
return err
|
|
}
|
|
return tx.Commit(ctx)
|
|
}
|
|
|
|
// HardDelete permanently removes the photo row. Bound objects are not touched
|
|
// here; the caller marks orphaned objects for collection.
|
|
func (s *Store) HardDelete(ctx context.Context, id string) error {
|
|
tag, err := s.pool.Exec(ctx, `DELETE FROM photos WHERE id = $1`, id)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if tag.RowsAffected() == 0 {
|
|
return ErrNotFound
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// ListTrash returns soft-deleted photos from albums the user belongs to, plus
|
|
// their own photos whose album no longer exists.
|
|
func (s *Store) ListTrash(ctx context.Context, userID string) ([]models.Photo, error) {
|
|
rows, err := s.pool.Query(ctx,
|
|
`SELECT `+photoColumns+`, COALESCE(a.title, '')
|
|
FROM photos p
|
|
JOIN users u ON u.id = p.uploader_id
|
|
LEFT JOIN albums a ON a.id = p.album_id`+photoJoins+`
|
|
WHERE p.deleted_at IS NOT NULL
|
|
AND (
|
|
(a.id IS NOT NULL AND (a.owner_id = $1 OR EXISTS(
|
|
SELECT 1 FROM album_members m WHERE m.album_id = a.id AND m.user_id = $1)))
|
|
OR (p.album_id IS NULL AND p.uploader_id = $1)
|
|
)
|
|
ORDER BY p.deleted_at DESC
|
|
LIMIT 200`,
|
|
userID, userID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
|
|
out := []models.Photo{}
|
|
for rows.Next() {
|
|
var p models.Photo
|
|
var exifText, tagsText string
|
|
if err := rows.Scan(&p.ID, &p.AlbumID, &p.UploaderID, &p.StorageKey,
|
|
&p.ThumbKey, &p.PreviewKey,
|
|
&p.OriginalFilename, &p.Size, &p.Width, &p.Height, &p.MimeType,
|
|
&p.TakenAt, &p.Latitude, &p.Longitude, &p.Caption, &p.CreatedAt,
|
|
&p.Uploader.ID, &p.Uploader.Name, &p.Uploader.AvatarURL,
|
|
&p.LikesCount, &p.CommentsCount, &p.LikedByMe, &exifText, &p.DeletedAt, &p.AlbumTitle, &tagsText); err != nil {
|
|
return nil, err
|
|
}
|
|
setExif(&p, exifText)
|
|
setTags(&p, tagsText)
|
|
out = append(out, p)
|
|
}
|
|
return out, rows.Err()
|
|
}
|
|
|
|
// TrashCount counts soft-deleted photos accessible to the user.
|
|
func (s *Store) TrashCount(ctx context.Context, userID string) (int, error) {
|
|
var n int
|
|
err := s.pool.QueryRow(ctx,
|
|
`SELECT count(*) FROM photos p
|
|
LEFT JOIN albums a ON a.id = p.album_id
|
|
WHERE p.deleted_at IS NOT NULL
|
|
AND (
|
|
(a.id IS NOT NULL AND (a.owner_id = $1 OR EXISTS(
|
|
SELECT 1 FROM album_members m WHERE m.album_id = a.id AND m.user_id = $1)))
|
|
OR (p.album_id IS NULL AND p.uploader_id = $1)
|
|
)`, userID).Scan(&n)
|
|
return n, err
|
|
}
|
|
|
|
// RestoreToAlbum reassigns a soft-deleted photo to an album and clears the trash flag.
|
|
func (s *Store) RestoreToAlbum(ctx context.Context, id, albumID string) error {
|
|
tag, err := s.pool.Exec(ctx,
|
|
`UPDATE photos SET album_id = $2, deleted_at = NULL WHERE id = $1`, id, albumID)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if tag.RowsAffected() == 0 {
|
|
return ErrNotFound
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// AlbumExists reports whether an album row exists.
|
|
func (s *Store) AlbumExists(ctx context.Context, albumID string) (bool, error) {
|
|
var ok bool
|
|
err := s.pool.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM albums WHERE id = $1)`, albumID).Scan(&ok)
|
|
return ok, err
|
|
}
|
|
|
|
func (s *Store) ListTags(ctx context.Context, userID, q string) ([]models.Tag, error) {
|
|
rows, err := s.pool.Query(ctx,
|
|
`SELECT id, name FROM tags
|
|
WHERE user_id = $1 AND ($2 = '' OR name ILIKE '%' || $2 || '%')
|
|
ORDER BY lower(name) LIMIT 50`, userID, q)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
out := []models.Tag{}
|
|
for rows.Next() {
|
|
var t models.Tag
|
|
if err := rows.Scan(&t.ID, &t.Name); err != nil {
|
|
return nil, err
|
|
}
|
|
out = append(out, t)
|
|
}
|
|
return out, rows.Err()
|
|
}
|
|
|
|
// EnsureTag returns the user's tag with the given name, creating it if needed.
|
|
func (s *Store) EnsureTag(ctx context.Context, userID, name string) (*models.Tag, error) {
|
|
name = strings.TrimSpace(name)
|
|
if name == "" || len(name) > 64 {
|
|
return nil, ErrNotFound
|
|
}
|
|
var t models.Tag
|
|
err := s.pool.QueryRow(ctx,
|
|
`SELECT id, name FROM tags WHERE user_id = $1 AND lower(name) = lower($2)`, userID, name).
|
|
Scan(&t.ID, &t.Name)
|
|
if err == nil {
|
|
return &t, nil
|
|
}
|
|
if !errors.Is(err, pgx.ErrNoRows) {
|
|
return nil, err
|
|
}
|
|
id := auth.NewID()
|
|
err = s.pool.QueryRow(ctx,
|
|
`INSERT INTO tags (id, user_id, name) VALUES ($1,$2,$3) RETURNING id, name`, id, userID, name).
|
|
Scan(&t.ID, &t.Name)
|
|
if err != nil {
|
|
if e2 := s.pool.QueryRow(ctx,
|
|
`SELECT id, name FROM tags WHERE user_id = $1 AND lower(name) = lower($2)`, userID, name).
|
|
Scan(&t.ID, &t.Name); e2 == nil {
|
|
return &t, nil
|
|
}
|
|
return nil, err
|
|
}
|
|
return &t, nil
|
|
}
|
|
|
|
func (s *Store) AttachTag(ctx context.Context, photoID, tagID string) error {
|
|
_, err := s.pool.Exec(ctx,
|
|
`INSERT INTO photo_tags (photo_id, tag_id) VALUES ($1,$2) ON CONFLICT DO NOTHING`, photoID, tagID)
|
|
return err
|
|
}
|
|
|
|
func (s *Store) DetachTag(ctx context.Context, photoID, tagID string) error {
|
|
_, err := s.pool.Exec(ctx,
|
|
`DELETE FROM photo_tags WHERE photo_id = $1 AND tag_id = $2`, photoID, tagID)
|
|
return err
|
|
}
|
|
|
|
func (s *Store) TagOwnedBy(ctx context.Context, tagID, userID string) (bool, error) {
|
|
var ok bool
|
|
err := s.pool.QueryRow(ctx,
|
|
`SELECT EXISTS(SELECT 1 FROM tags WHERE id = $1 AND user_id = $2)`, tagID, userID).Scan(&ok)
|
|
return ok, err
|
|
}
|
|
|
|
func (s *Store) SetCoverIfEmpty(ctx context.Context, albumID, photoID string) {
|
|
_, _ = s.pool.Exec(ctx,
|
|
`UPDATE albums SET cover_photo_id = $2 WHERE id = $1 AND cover_photo_id IS NULL`,
|
|
albumID, photoID)
|
|
}
|
|
|
|
func (s *Store) UpdateFilename(ctx context.Context, id, filename string) error {
|
|
tag, err := s.pool.Exec(ctx, `UPDATE photos SET original_filename = $2 WHERE id = $1`, id, filename)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if tag.RowsAffected() == 0 {
|
|
return ErrNotFound
|
|
}
|
|
return nil
|
|
}
|
|
|
|
type backfillRow struct {
|
|
ID string
|
|
StorageKey string
|
|
}
|
|
|
|
// BackfillEXIF re-reads the original object of every photo with empty exif and
|
|
// fills the exif/geo fields on the photo.
|
|
func (s *Store) BackfillEXIF(ctx context.Context, fetch func(context.Context, string) ([]byte, error)) (int, error) {
|
|
rows, err := s.pool.Query(ctx,
|
|
`SELECT p.id, o.storage_key
|
|
FROM photos p
|
|
JOIN photo_objects po ON po.photo_id = p.id AND po.role = 'original'
|
|
JOIN objects o ON o.id = po.object_id
|
|
WHERE p.exif IS NULL OR p.exif = '{}'::jsonb
|
|
ORDER BY p.created_at`)
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
var pending []backfillRow
|
|
for rows.Next() {
|
|
var r backfillRow
|
|
if err := rows.Scan(&r.ID, &r.StorageKey); err != nil {
|
|
rows.Close()
|
|
return 0, err
|
|
}
|
|
pending = append(pending, r)
|
|
}
|
|
rows.Close()
|
|
if err := rows.Err(); err != nil {
|
|
return 0, err
|
|
}
|
|
|
|
updated := 0
|
|
for _, r := range pending {
|
|
data, err := fetch(ctx, r.StorageKey)
|
|
if err != nil {
|
|
continue
|
|
}
|
|
meta := extractEXIF(data)
|
|
if len(meta.Attributes) == 0 && meta.TakenAt == nil && meta.Latitude == nil && meta.Longitude == nil {
|
|
continue
|
|
}
|
|
attrs, err := json.Marshal(meta.Attributes)
|
|
if err != nil || len(attrs) == 0 {
|
|
attrs = []byte("{}")
|
|
}
|
|
_, err = s.pool.Exec(ctx,
|
|
`UPDATE photos SET exif = $2::jsonb,
|
|
taken_at = COALESCE(taken_at, $3),
|
|
latitude = COALESCE(latitude, $4),
|
|
longitude = COALESCE(longitude, $5)
|
|
WHERE id = $1`,
|
|
r.ID, string(attrs), meta.TakenAt, meta.Latitude, meta.Longitude)
|
|
if err != nil {
|
|
continue
|
|
}
|
|
updated++
|
|
}
|
|
return updated, nil
|
|
}
|