bsdaily/internal/bsdaily/extract.go

package bsdaily

import (
	"database/sql"
	"errors"
	"fmt"
	"log/slog"
	"time"

	_ "modernc.org/sqlite"
)

var ErrNoPosts = errors.New("no posts found for target day")

// ExtractDay opens a new empty database at dstDBPath, attaches srcDBPath,
// and copies only the target day's data into it. This is much faster than
// pruning a full copy because it only reads/writes the small slice of data
// being kept.
func ExtractDay(srcDBPath, dstDBPath string, targetDay time.Time) error {
	dayStart := targetDay.Format("2006-01-02") + "T00:00:00"
	dayEnd := targetDay.AddDate(0, 0, 1).Format("2006-01-02") + "T00:00:00"

	slog.Info("extracting day", "from", dayStart, "until", dayEnd)

	// Maximum performance pragmas - we don't care about crash safety for temp files
	// Use WAL mode for the source attachment to avoid locking issues
	pragmas := fmt.Sprintf("?_pragma=journal_mode(WAL)&_pragma=synchronous(OFF)&_pragma=cache_size(%d)&_pragma=foreign_keys(OFF)&_pragma=temp_store(MEMORY)&_pragma=busy_timeout(5000)", sqliteCacheSizeKB)
	db, err := sql.Open("sqlite", dstDBPath+pragmas)
	if err != nil {
		return fmt.Errorf("opening destination database: %w", err)
	}
	defer db.Close()

	// Attach source database
	if _, err := db.Exec("ATTACH DATABASE ? AS src", srcDBPath); err != nil {
		return fmt.Errorf("attaching source database: %w", err)
	}

	// Copy table DDL from source
	slog.Info("copying table DDL from source")
	rows, err := db.Query("SELECT sql FROM src.sqlite_master WHERE type='table' AND name NOT LIKE 'sqlite_%' ORDER BY name")
	if err != nil {
		return fmt.Errorf("reading source schema: %w", err)
	}
	defer rows.Close()

	var ddlStatements []string
	for rows.Next() {
		var ddl string
		if err := rows.Scan(&ddl); err != nil {
			return fmt.Errorf("scanning DDL: %w", err)
		}
		ddlStatements = append(ddlStatements, ddl)
	}
	if err := rows.Err(); err != nil {
		return fmt.Errorf("iterating DDL rows: %w", err)
	}

	for _, ddl := range ddlStatements {
		if _, err := db.Exec(ddl); err != nil {
			return fmt.Errorf("creating table: %w\nDDL: %s", err, ddl)
		}
	}

	// Begin transaction for bulk inserts
	tx, err := db.Begin()
	if err != nil {
		return fmt.Errorf("beginning transaction: %w", err)
	}
	defer func() {
		if err != nil {
			tx.Rollback()
		}
	}()

	// Insert target day's data
	slog.Info("inserting posts for target day")
	result, err := tx.Exec("INSERT INTO posts SELECT * FROM src.posts WHERE timestamp >= ? AND timestamp < ?", dayStart, dayEnd)
	if err != nil {
		return fmt.Errorf("inserting posts: %w", err)
	}
	postCount, _ := result.RowsAffected()
	slog.Info("inserted posts", "count", postCount)

	if postCount == 0 {
		return fmt.Errorf("%w %s - aborting to avoid producing empty output",
			ErrNoPosts, targetDay.Format("2006-01-02"))
	}

	slog.Info("inserting junction and lookup tables")
	if _, err := tx.Exec("INSERT INTO posts_hashtags SELECT * FROM src.posts_hashtags WHERE post_id IN (SELECT id FROM posts)"); err != nil {
		return fmt.Errorf("inserting posts_hashtags: %w", err)
	}

	if _, err := tx.Exec("INSERT INTO posts_urls SELECT * FROM src.posts_urls WHERE post_id IN (SELECT id FROM posts)"); err != nil {
		return fmt.Errorf("inserting posts_urls: %w", err)
	}

	if _, err := tx.Exec("INSERT INTO hashtags SELECT * FROM src.hashtags WHERE id IN (SELECT hashtag_id FROM posts_hashtags)"); err != nil {
		return fmt.Errorf("inserting hashtags: %w", err)
	}

	if _, err := tx.Exec("INSERT INTO urls SELECT * FROM src.urls WHERE id IN (SELECT url_id FROM posts_urls)"); err != nil {
		return fmt.Errorf("inserting urls: %w", err)
	}

	if _, err := tx.Exec("INSERT INTO users SELECT * FROM src.users WHERE did IN (SELECT user_did FROM posts)"); err != nil {
		return fmt.Errorf("inserting users: %w", err)
	}

	// Check if media table exists in source and copy if present
	var mediaTableExists int
	if err := tx.QueryRow("SELECT COUNT(*) FROM src.sqlite_master WHERE type='table' AND name='media'").Scan(&mediaTableExists); err != nil {
		slog.Warn("checking for media table", "error", err)
	} else if mediaTableExists > 0 {
		slog.Info("inserting media entries")
		// Get post blob_cids for this day's posts
		if _, err := tx.Exec("INSERT INTO media SELECT * FROM src.media WHERE content_hash IN (SELECT blob_cids FROM posts WHERE blob_cids IS NOT NULL)"); err != nil {
			slog.Warn("inserting media (may not have matching entries)", "error", err)
		}
	}

	// Commit the transaction before any further database operations
	if err := tx.Commit(); err != nil {
		return fmt.Errorf("committing transaction: %w", err)
	}
	tx = nil // Clear tx to ensure defer doesn't try to rollback

	// Create indexes after bulk insert for speed
	slog.Info("creating indexes")
	idxRows, err := db.Query("SELECT sql FROM src.sqlite_master WHERE type='index' AND name NOT LIKE 'sqlite_%' AND sql IS NOT NULL ORDER BY name")
	if err != nil {
		return fmt.Errorf("reading source indexes: %w", err)
	}
	defer idxRows.Close()

	var idxStatements []string
	for idxRows.Next() {
		var idxSQL string
		if err := idxRows.Scan(&idxSQL); err != nil {
			return fmt.Errorf("scanning index DDL: %w", err)
		}
		idxStatements = append(idxStatements, idxSQL)
	}
	if err := idxRows.Err(); err != nil {
		return fmt.Errorf("iterating index rows: %w", err)
	}

	for _, idxSQL := range idxStatements {
		if _, err := db.Exec(idxSQL); err != nil {
			return fmt.Errorf("creating index: %w\nDDL: %s", err, idxSQL)
		}
	}

	// Detach source
	if _, err := db.Exec("DETACH DATABASE src"); err != nil {
		return fmt.Errorf("detaching source database: %w", err)
	}

	// Verify post count
	var verifyCount int64
	if err := db.QueryRow("SELECT COUNT(*) FROM posts").Scan(&verifyCount); err != nil {
		return fmt.Errorf("verifying post count: %w", err)
	}
	if verifyCount != postCount {
		return fmt.Errorf("post count mismatch: inserted %d but found %d", postCount, verifyCount)
	}
	slog.Info("extraction complete", "posts", verifyCount)

	return nil
}