Implement local SQLite index database with repositories

- Add SQLite database connection management with proper error handling - Implement schema for files, chunks, blobs, and snapshots tables - Create repository pattern for each database table - Add transaction support with proper rollback handling - Integrate database module with fx dependency injection - Make index path configurable via VAULTIK_INDEX_PATH env var - Add fatal error handling for database integrity issues - Update DESIGN.md to clarify file_chunks vs chunk_files distinction - Remove FinalHash from BlobInfo (blobs are content-addressable) - Add file metadata support (mtime, ctime, mode, uid, gid, symlinks)
2025-07-20 10:26:15 +02:00
parent 9de439a0a4
commit b2e85d9e76
31 changed files with 1229 additions and 83 deletions
--- a/internal/database/chunks.go
+++ b/internal/database/chunks.go
@@ -0,0 +1,141 @@
+package database
+
+import (
+	"context"
+	"database/sql"
+	"fmt"
+)
+
+type ChunkRepository struct {
+	db *DB
+}
+
+func NewChunkRepository(db *DB) *ChunkRepository {
+	return &ChunkRepository{db: db}
+}
+
+func (r *ChunkRepository) Create(ctx context.Context, tx *sql.Tx, chunk *Chunk) error {
+	query := `
+		INSERT INTO chunks (chunk_hash, sha256, size)
+		VALUES (?, ?, ?)
+		ON CONFLICT(chunk_hash) DO NOTHING
+	`
+
+	var err error
+	if tx != nil {
+		_, err = tx.ExecContext(ctx, query, chunk.ChunkHash, chunk.SHA256, chunk.Size)
+	} else {
+		_, err = r.db.conn.ExecContext(ctx, query, chunk.ChunkHash, chunk.SHA256, chunk.Size)
+	}
+
+	if err != nil {
+		return fmt.Errorf("inserting chunk: %w", err)
+	}
+
+	return nil
+}
+
+func (r *ChunkRepository) GetByHash(ctx context.Context, hash string) (*Chunk, error) {
+	query := `
+		SELECT chunk_hash, sha256, size
+		FROM chunks
+		WHERE chunk_hash = ?
+	`
+
+	var chunk Chunk
+
+	err := r.db.conn.QueryRowContext(ctx, query, hash).Scan(
+		&chunk.ChunkHash,
+		&chunk.SHA256,
+		&chunk.Size,
+	)
+
+	if err == sql.ErrNoRows {
+		return nil, nil
+	}
+	if err != nil {
+		return nil, fmt.Errorf("querying chunk: %w", err)
+	}
+
+	return &chunk, nil
+}
+
+func (r *ChunkRepository) GetByHashes(ctx context.Context, hashes []string) ([]*Chunk, error) {
+	if len(hashes) == 0 {
+		return nil, nil
+	}
+
+	query := `
+		SELECT chunk_hash, sha256, size
+		FROM chunks
+		WHERE chunk_hash IN (`
+
+	args := make([]interface{}, len(hashes))
+	for i, hash := range hashes {
+		if i > 0 {
+			query += ", "
+		}
+		query += "?"
+		args[i] = hash
+	}
+	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 CloseRows(rows)
+
+	var chunks []*Chunk
+	for rows.Next() {
+		var chunk Chunk
+
+		err := rows.Scan(
+			&chunk.ChunkHash,
+			&chunk.SHA256,
+			&chunk.Size,
+		)
+		if err != nil {
+			return nil, fmt.Errorf("scanning chunk: %w", err)
+		}
+
+		chunks = append(chunks, &chunk)
+	}
+
+	return chunks, rows.Err()
+}
+
+func (r *ChunkRepository) ListUnpacked(ctx context.Context, limit int) ([]*Chunk, error) {
+	query := `
+		SELECT c.chunk_hash, c.sha256, 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 CloseRows(rows)
+
+	var chunks []*Chunk
+	for rows.Next() {
+		var chunk Chunk
+
+		err := rows.Scan(
+			&chunk.ChunkHash,
+			&chunk.SHA256,
+			&chunk.Size,
+		)
+		if err != nil {
+			return nil, fmt.Errorf("scanning chunk: %w", err)
+		}
+
+		chunks = append(chunks, &chunk)
+	}
+
+	return chunks, rows.Err()
+}