vaultik/internal/database/blobs.go

package database

import (
	"context"
	"database/sql"
	"errors"
	"fmt"
	"time"

	"sneak.berlin/go/vaultik/internal/log"
)

// BlobRepository provides access to the blobs table, which tracks the
// packed, encrypted storage units uploaded to the destination.
type BlobRepository struct {
	db *DB
}

// NewBlobRepository creates a BlobRepository backed by db.
func NewBlobRepository(db *DB) *BlobRepository {
	return &BlobRepository{db: db}
}

// Create inserts a blob row, using tx when non-nil.
func (r *BlobRepository) Create(ctx context.Context, tx *sql.Tx, blob *Blob) error {
	query := `
		INSERT INTO blobs (id, blob_hash, created_ts, finished_ts,
			uncompressed_size, compressed_size, uploaded_ts)
		VALUES (?, ?, ?, ?, ?, ?, ?)
	`

	var finishedTS, uploadedTS *int64

	if blob.FinishedTS != nil {
		ts := blob.FinishedTS.Unix()
		finishedTS = &ts
	}

	if blob.UploadedTS != nil {
		ts := blob.UploadedTS.Unix()
		uploadedTS = &ts
	}

	var err error
	if tx != nil {
		_, err = tx.ExecContext(ctx, query, blob.ID, blob.Hash, blob.CreatedTS.Unix(),
			finishedTS, blob.UncompressedSize, blob.CompressedSize, uploadedTS)
	} else {
		_, err = r.db.ExecWithLog(ctx, query, blob.ID, blob.Hash, blob.CreatedTS.Unix(),
			finishedTS, blob.UncompressedSize, blob.CompressedSize, uploadedTS)
	}

	if err != nil {
		return fmt.Errorf("inserting blob: %w", err)
	}

	return nil
}

// GetByHash returns the blob with the given content hash, or nil if no
// such blob exists.
func (r *BlobRepository) GetByHash(ctx context.Context, hash string) (*Blob, error) {
	return r.getOne(ctx, "blob_hash", hash)
}

// GetByID retrieves a blob by its ID
func (r *BlobRepository) GetByID(ctx context.Context, id string) (*Blob, error) {
	return r.getOne(ctx, "id", id)
}

// GetAll returns every blob row keyed by blob ID. Useful at restore
// start to translate the per-chunk blob_id references in chunkToBlobMap
// into blob hashes without doing one GetByID query per chunk.
func (r *BlobRepository) GetAll(ctx context.Context) (map[string]*Blob, error) {
	query := `
		SELECT id, blob_hash, created_ts, finished_ts,
			uncompressed_size, compressed_size, uploaded_ts
		FROM blobs
	`

	rows, err := r.db.conn.QueryContext(ctx, query)
	if err != nil {
		return nil, fmt.Errorf("querying blobs: %w", err)
	}

	defer func() { _ = rows.Close() }()

	out := make(map[string]*Blob)

	for rows.Next() {
		var (
			blob                           Blob
			createdTSUnix                  int64
			finishedTSUnix, uploadedTSUnix sql.NullInt64
		)

		err := rows.Scan(
			&blob.ID,
			&blob.Hash,
			&createdTSUnix,
			&finishedTSUnix,
			&blob.UncompressedSize,
			&blob.CompressedSize,
			&uploadedTSUnix,
		)
		if err != nil {
			return nil, fmt.Errorf("scanning blob: %w", err)
		}

		blob.CreatedTS = time.Unix(createdTSUnix, 0).UTC()
		if finishedTSUnix.Valid {
			ts := time.Unix(finishedTSUnix.Int64, 0).UTC()
			blob.FinishedTS = &ts
		}

		if uploadedTSUnix.Valid {
			ts := time.Unix(uploadedTSUnix.Int64, 0).UTC()
			blob.UploadedTS = &ts
		}

		out[blob.ID.String()] = &blob
	}

	return out, rows.Err()
}

// UpdateFinished updates a blob when it's finalized
func (r *BlobRepository) UpdateFinished(
	ctx context.Context,
	tx *sql.Tx,
	id string,
	hash string,
	uncompressedSize, compressedSize int64,
) error {
	query := `
		UPDATE blobs 
		SET blob_hash = ?, finished_ts = ?, uncompressed_size = ?, compressed_size = ?
		WHERE id = ?
	`

	now := time.Now().UTC().Unix()

	var err error
	if tx != nil {
		_, err = tx.ExecContext(ctx, query, hash, now, uncompressedSize, compressedSize, id)
	} else {
		_, err = r.db.ExecWithLog(ctx, query, hash, now, uncompressedSize, compressedSize, id)
	}

	if err != nil {
		return fmt.Errorf("updating blob: %w", err)
	}

	return nil
}

// UpdateUploaded marks a blob as uploaded and records the size of the
// stored object as its compressed size. That size can differ from the
// local copy's when an identical blob was already stored with another
// compression level or other recipients.
func (r *BlobRepository) UpdateUploaded(
	ctx context.Context, tx *sql.Tx, id string, storedSize int64,
) error {
	query := `
		UPDATE blobs
		SET uploaded_ts = ?, compressed_size = ?
		WHERE id = ?
	`

	now := time.Now().UTC().Unix()

	var err error
	if tx != nil {
		_, err = tx.ExecContext(ctx, query, now, storedSize, id)
	} else {
		_, err = r.db.ExecWithLog(ctx, query, now, storedSize, id)
	}

	if err != nil {
		return fmt.Errorf("marking blob as uploaded: %w", err)
	}

	return nil
}

// DeleteOrphaned deletes blobs that are not referenced by any snapshot
func (r *BlobRepository) DeleteOrphaned(ctx context.Context) error {
	query := `
		DELETE FROM blobs 
		WHERE NOT EXISTS (
			SELECT 1 FROM snapshot_blobs 
			WHERE snapshot_blobs.blob_id = blobs.id
		)
	`

	result, err := r.db.ExecWithLog(ctx, query)
	if err != nil {
		return fmt.Errorf("deleting orphaned blobs: %w", err)
	}

	rowsAffected, _ := result.RowsAffected()
	if rowsAffected > 0 {
		log.Debug("Deleted orphaned blobs", "count", rowsAffected)
	}

	return nil
}

// DeleteUnuploaded deletes blob rows whose upload never completed
// (uploaded_ts IS NULL) and returns how many were removed. Their
// blob_chunks rows are removed by the ON DELETE CASCADE foreign key.
// A blob is only ever attached to a snapshot once its upload has been
// recorded, so an un-uploaded blob is never referenced by a completed
// snapshot: dropping it discards chunk rows that point at data which
// was never stored remotely, so the affected content is re-chunked and
// re-uploaded on the next run.
func (r *BlobRepository) DeleteUnuploaded(ctx context.Context) (int64, error) {
	query := `DELETE FROM blobs WHERE uploaded_ts IS NULL`

	result, err := r.db.ExecWithLog(ctx, query)
	if err != nil {
		return 0, fmt.Errorf("deleting un-uploaded blobs: %w", err)
	}

	rowsAffected, _ := result.RowsAffected()
	if rowsAffected > 0 {
		log.Debug("Deleted un-uploaded blobs", "count", rowsAffected)
	}

	return rowsAffected, nil
}

// getOne fetches a single blob row matched on the given column, or
// (nil, nil) when no row matches.
func (r *BlobRepository) getOne(
	ctx context.Context, column, value string,
) (*Blob, error) {
	query := `
		SELECT id, blob_hash, created_ts, finished_ts,
			uncompressed_size, compressed_size, uploaded_ts
		FROM blobs
		WHERE ` + column + ` = ?`

	var (
		blob                           Blob
		createdTSUnix                  int64
		finishedTSUnix, uploadedTSUnix sql.NullInt64
	)

	err := r.db.conn.QueryRowContext(ctx, query, value).Scan(
		&blob.ID,
		&blob.Hash,
		&createdTSUnix,
		&finishedTSUnix,
		&blob.UncompressedSize,
		&blob.CompressedSize,
		&uploadedTSUnix,
	)

	if errors.Is(err, sql.ErrNoRows) {
		return nil, nil //nolint:nilnil // nil,nil signals not-found; callers check nil
	}

	if err != nil {
		return nil, fmt.Errorf("querying blob: %w", err)
	}

	blob.CreatedTS = time.Unix(createdTSUnix, 0).UTC()
	if finishedTSUnix.Valid {
		ts := time.Unix(finishedTSUnix.Int64, 0).UTC()
		blob.FinishedTS = &ts
	}

	if uploadedTSUnix.Valid {
		ts := time.Unix(uploadedTSUnix.Int64, 0).UTC()
		blob.UploadedTS = &ts
	}

	return &blob, nil
}