mcr/internal/db/push.go

package db

import (
	"database/sql"
	"errors"
	"fmt"
)

// GetOrCreateRepository returns the repository ID for the given name,
// creating it if it does not exist (implicit creation on first push).
func (d *DB) GetOrCreateRepository(name string) (int64, error) {
	var id int64
	err := d.QueryRow(`SELECT id FROM repositories WHERE name = ?`, name).Scan(&id)
	if err == nil {
		return id, nil
	}
	if !errors.Is(err, sql.ErrNoRows) {
		return 0, fmt.Errorf("db: get repository: %w", err)
	}

	result, err := d.Exec(`INSERT INTO repositories (name) VALUES (?)`, name)
	if err != nil {
		return 0, fmt.Errorf("db: create repository: %w", err)
	}
	id, err = result.LastInsertId()
	if err != nil {
		return 0, fmt.Errorf("db: repository last insert id: %w", err)
	}
	return id, nil
}

// BlobExists checks whether a blob with the given digest exists in the blobs table.
func (d *DB) BlobExists(digest string) (bool, error) {
	var count int
	err := d.QueryRow(`SELECT COUNT(*) FROM blobs WHERE digest = ?`, digest).Scan(&count)
	if err != nil {
		return false, fmt.Errorf("db: blob exists: %w", err)
	}
	return count > 0, nil
}

// InsertBlob inserts a blob row if it does not already exist.
// Returns without error if the blob already exists (content-addressed dedup).
func (d *DB) InsertBlob(digest string, size int64) error {
	_, err := d.Exec(
		`INSERT OR IGNORE INTO blobs (digest, size) VALUES (?, ?)`,
		digest, size,
	)
	if err != nil {
		return fmt.Errorf("db: insert blob: %w", err)
	}
	return nil
}

// PushManifestParams holds the parameters for a manifest push operation.
type PushManifestParams struct {
	RepoName    string
	Digest      string
	MediaType   string
	Content     []byte
	Size        int64
	Tag         string   // empty if push-by-digest
	BlobDigests []string // referenced blob digests
}

// PushManifest executes the full manifest push in a single transaction per
// ARCHITECTURE.md §5. It creates the repository if needed, inserts/updates
// the manifest, populates manifest_blobs, and updates the tag if provided.
func (d *DB) PushManifest(p PushManifestParams) error {
	tx, err := d.Begin()
	if err != nil {
		return fmt.Errorf("db: begin push manifest: %w", err)
	}

	// Step a: create repository if not exists.
	var repoID int64
	err = tx.QueryRow(`SELECT id FROM repositories WHERE name = ?`, p.RepoName).Scan(&repoID)
	if errors.Is(err, sql.ErrNoRows) {
		result, insertErr := tx.Exec(`INSERT INTO repositories (name) VALUES (?)`, p.RepoName)
		if insertErr != nil {
			_ = tx.Rollback()
			return fmt.Errorf("db: create repository: %w", insertErr)
		}
		repoID, err = result.LastInsertId()
		if err != nil {
			_ = tx.Rollback()
			return fmt.Errorf("db: repository last insert id: %w", err)
		}
	} else if err != nil {
		_ = tx.Rollback()
		return fmt.Errorf("db: get repository: %w", err)
	}

	// Step b: insert or update manifest.
	// Use INSERT ... ON CONFLICT DO UPDATE on the UNIQUE(repository_id, digest) constraint.
	_, err = tx.Exec(
		`INSERT INTO manifests (repository_id, digest, media_type, content, size)
		 VALUES (?, ?, ?, ?, ?)
		 ON CONFLICT(repository_id, digest) DO UPDATE SET
		     media_type = excluded.media_type,
		     content = excluded.content,
		     size = excluded.size`,
		repoID, p.Digest, p.MediaType, p.Content, p.Size,
	)
	if err != nil {
		_ = tx.Rollback()
		return fmt.Errorf("db: insert manifest: %w", err)
	}

	// Retrieve the manifest ID by querying the row directly. We cannot use
	// result.LastInsertId() here because SQLite's last_insert_rowid() is
	// unreliable after an ON CONFLICT DO UPDATE — it returns the rowid of
	// the most recent *insert* in the connection, not the upserted row.
	// When the conflict branch fires (no new row inserted), the stale
	// last_insert_rowid from a previous insert is returned, causing
	// manifest_blobs and tags to reference the wrong manifest.
	var manifestID int64
	err = tx.QueryRow(
		`SELECT id FROM manifests WHERE repository_id = ? AND digest = ?`,
		repoID, p.Digest,
	).Scan(&manifestID)
	if err != nil {
		_ = tx.Rollback()
		return fmt.Errorf("db: get manifest id after upsert: %w", err)
	}

	// Step c: populate manifest_blobs join table.
	// Delete existing entries for this manifest first (in case of re-push).
	_, err = tx.Exec(`DELETE FROM manifest_blobs WHERE manifest_id = ?`, manifestID)
	if err != nil {
		_ = tx.Rollback()
		return fmt.Errorf("db: clear manifest_blobs: %w", err)
	}
	for _, blobDigest := range p.BlobDigests {
		_, err = tx.Exec(
			`INSERT INTO manifest_blobs (manifest_id, blob_id)
			 SELECT ?, id FROM blobs WHERE digest = ?`,
			manifestID, blobDigest,
		)
		if err != nil {
			_ = tx.Rollback()
			return fmt.Errorf("db: insert manifest_blob: %w", err)
		}
	}

	// Step d: if reference is a tag, insert or update tag row.
	if p.Tag != "" {
		_, err = tx.Exec(
			`INSERT INTO tags (repository_id, name, manifest_id)
			 VALUES (?, ?, ?)
			 ON CONFLICT(repository_id, name) DO UPDATE SET
			     manifest_id = excluded.manifest_id,
			     updated_at = strftime('%Y-%m-%dT%H:%M:%SZ','now')`,
			repoID, p.Tag, manifestID,
		)
		if err != nil {
			_ = tx.Rollback()
			return fmt.Errorf("db: upsert tag: %w", err)
		}
	}

	if err := tx.Commit(); err != nil {
		return fmt.Errorf("db: commit push manifest: %w", err)
	}
	return nil
}