vaultik/internal/database/chunks.go

package database

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

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

// ChunkRepository provides access to the chunks table, which tracks
// content-defined chunks by hash and size.
type ChunkRepository struct {
	db *DB
}

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

// Create inserts a chunk row (idempotently), using tx when non-nil.
func (r *ChunkRepository) Create(ctx context.Context, tx *sql.Tx, chunk *Chunk) error {
	query := `
		INSERT INTO chunks (chunk_hash, size)
		VALUES (?, ?)
		ON CONFLICT(chunk_hash) DO NOTHING
	`

	var err error
	if tx != nil {
		_, err = tx.ExecContext(ctx, query, chunk.ChunkHash, chunk.Size)
	} else {
		_, err = r.db.ExecWithLog(ctx, query, chunk.ChunkHash, chunk.Size)
	}

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

	return nil
}

// GetByHash returns the chunk with the given hash, or nil if it is not
// known to the index.
func (r *ChunkRepository) GetByHash(ctx context.Context, hash string) (*Chunk, error) {
	query := `
		SELECT chunk_hash, size
		FROM chunks
		WHERE chunk_hash = ?
	`

	var chunk Chunk

	err := r.db.conn.QueryRowContext(ctx, query, hash).Scan(
		&chunk.ChunkHash,
		&chunk.Size,
	)

	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 chunk: %w", err)
	}

	return &chunk, nil
}

// GetByHashes returns the chunks whose hashes appear in hashes, ordered by
// chunk hash. Unknown hashes are silently omitted from the result.
func (r *ChunkRepository) GetByHashes(
	ctx context.Context, hashes []string,
) ([]*Chunk, error) {
	if len(hashes) == 0 {
		return nil, nil
	}

	query := `
		SELECT chunk_hash, size
		FROM chunks
		WHERE chunk_hash IN (`

	args := make([]any, len(hashes))

	var querySb75 strings.Builder

	for i, hash := range hashes {
		if i > 0 {
			querySb75.WriteString(", ")
		}

		querySb75.WriteString("?")

		args[i] = hash
	}

	query += querySb75.String()

	query += ") ORDER BY chunk_hash"

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

	defer func() {
		err := rows.Close()
		if err != nil {
			Fatalf("failed to close rows: %v", err)
		}
	}()

	var chunks []*Chunk

	for rows.Next() {
		var chunk Chunk

		err := rows.Scan(
			&chunk.ChunkHash,
			&chunk.Size,
		)
		if err != nil {
			return nil, fmt.Errorf("scanning chunk: %w", err)
		}

		chunks = append(chunks, &chunk)
	}

	return chunks, rows.Err()
}

// ListUnpacked returns up to limit chunks that are not yet stored in any
// blob, ordered by chunk hash.
func (r *ChunkRepository) ListUnpacked(
	ctx context.Context, limit int,
) ([]*Chunk, error) {
	query := `
		SELECT c.chunk_hash, c.size
		FROM chunks c
		LEFT JOIN blob_chunks bc ON c.chunk_hash = bc.chunk_hash
		WHERE bc.chunk_hash IS NULL
		ORDER BY c.chunk_hash
		LIMIT ?
	`

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

	defer func() {
		err := rows.Close()
		if err != nil {
			Fatalf("failed to close rows: %v", err)
		}
	}()

	var chunks []*Chunk

	for rows.Next() {
		var chunk Chunk

		err := rows.Scan(
			&chunk.ChunkHash,
			&chunk.Size,
		)
		if err != nil {
			return nil, fmt.Errorf("scanning chunk: %w", err)
		}

		chunks = append(chunks, &chunk)
	}

	return chunks, rows.Err()
}

// DeleteOrphaned deletes chunks that are not referenced by any file or blob
func (r *ChunkRepository) DeleteOrphaned(ctx context.Context) error {
	query := `
		DELETE FROM chunks 
		WHERE NOT EXISTS (
			SELECT 1 FROM file_chunks 
			WHERE file_chunks.chunk_hash = chunks.chunk_hash
		)
		AND NOT EXISTS (
			SELECT 1 FROM blob_chunks 
			WHERE blob_chunks.chunk_hash = chunks.chunk_hash
		)
	`

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

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

	return nil
}