bsdaily/internal/bsdaily/run.go

package bsdaily

import (
	"errors"
	"fmt"
	"log/slog"
	"os"
	"path/filepath"
	"time"
)

// cleanup removes a temporary file, logging a warning if removal fails so
// that leaked scratch files are surfaced rather than silently ignored. A
// missing file is not an error.
func cleanup(path string) {
	if err := os.Remove(path); err != nil && !os.IsNotExist(err) {
		slog.Warn("failed to remove temporary file", "path", path, "error", err)
	}
}

func Run(targetDates []time.Time) error {
	snapshotDir, snapshotDate, err := FindLatestDailySnapshot()
	if err != nil {
		return fmt.Errorf("finding latest snapshot: %w", err)
	}
	slog.Info("found latest daily snapshot", "dir", snapshotDir, "snapshot_date", snapshotDate.Format("2006-01-02"))

	if len(targetDates) == 0 {
		targetDates = []time.Time{snapshotDate.AddDate(0, 0, -1)}
	}

	slog.Info("target days for extraction", "count", len(targetDates),
		"first", targetDates[0].Format("2006-01-02"),
		"last", targetDates[len(targetDates)-1].Format("2006-01-02"))

	// Check disk space
	if err := CheckFreeSpace(TmpBase, MinTmpFreeBytes, "tmpBase"); err != nil {
		return err
	}
	if err := CheckFreeSpace(DailiesBase, MinDailiesFreeBytes, "dailiesBase"); err != nil {
		return err
	}

	// Create temp directory
	tmpDir, err := os.MkdirTemp(TmpBase, "bsarchivesegment-*")
	if err != nil {
		return fmt.Errorf("creating temp directory in %s: %w", TmpBase, err)
	}
	slog.Info("created temp directory", "path", tmpDir)
	defer func() {
		slog.Info("cleaning up temp directory", "path", tmpDir)
		if err := os.RemoveAll(tmpDir); err != nil {
			slog.Error("failed to remove temp directory", "path", tmpDir, "error", err)
		}
	}()

	// Copy database files from snapshot to temp
	srcDB := filepath.Join(snapshotDir, DBFilename)
	srcWAL := filepath.Join(snapshotDir, WALFilename)
	srcSHM := filepath.Join(snapshotDir, SHMFilename)
	dstDB := filepath.Join(tmpDir, DBFilename)
	dstWAL := filepath.Join(tmpDir, WALFilename)
	dstSHM := filepath.Join(tmpDir, SHMFilename)

	for _, f := range []string{srcDB, srcWAL} {
		info, err := os.Stat(f)
		if err != nil {
			return fmt.Errorf("source file missing: %s: %w", f, err)
		}
		if info.Size() == 0 {
			return fmt.Errorf("source file is empty: %s", f)
		}
		slog.Info("source file", "path", f, "size_bytes", info.Size())
	}

	if err := CopyFile(srcDB, dstDB); err != nil {
		return fmt.Errorf("copying database: %w", err)
	}
	if err := CopyFile(srcWAL, dstWAL); err != nil {
		return fmt.Errorf("copying WAL: %w", err)
	}
	if _, err := os.Stat(srcSHM); err == nil {
		if err := CopyFile(srcSHM, dstSHM); err != nil {
			return fmt.Errorf("copying SHM: %w", err)
		}
	}

	// Process each day completely before moving to the next
	// This ensures we don't have multiple SQLite operations competing for the same source database
	processed := 0
	skipped := 0

	for _, targetDay := range targetDates {
		dayStr := targetDay.Format("2006-01-02")
		slog.Info("processing day", "date", dayStr)

		// Check if output already exists
		outputDir := filepath.Join(DailiesBase, targetDay.Format("2006-01"))
		outputFinal := filepath.Join(outputDir, dayStr+".sql.zst")
		if _, err := os.Stat(outputFinal); err == nil {
			slog.Info("output already exists, skipping", "path", outputFinal)
			skipped++
			continue
		}

		// Extract target day into a per-day database
		extractedDB := filepath.Join(tmpDir, "extracted-"+dayStr+".db")
		slog.Info("extracting target day", "src", dstDB, "dst", extractedDB)
		if err := ExtractDay(dstDB, extractedDB, targetDay); err != nil {
			if errors.Is(err, ErrNoPosts) {
				slog.Warn("no posts found, skipping day", "date", dayStr)
				cleanup(extractedDB)
				skipped++
				continue
			}
			return fmt.Errorf("extracting day %s: %w", dayStr, err)
		}

		// Dump to SQL and compress
		if err := os.MkdirAll(outputDir, 0755); err != nil {
			cleanup(extractedDB)
			return fmt.Errorf("creating output directory %s: %w", outputDir, err)
		}

		outputTmp := filepath.Join(outputDir, "."+dayStr+".sql.zst.tmp")

		slog.Info("dumping and compressing", "tmp_output", outputTmp)
		if err := DumpAndCompress(extractedDB, outputTmp); err != nil {
			cleanup(outputTmp)
			cleanup(extractedDB)
			return fmt.Errorf("dump and compress for %s: %w", dayStr, err)
		}

		slog.Info("verifying compressed output")
		if err := VerifyOutput(outputTmp); err != nil {
			cleanup(outputTmp)
			cleanup(extractedDB)
			return fmt.Errorf("verification failed for %s: %w", dayStr, err)
		}

		// Atomic rename to final path
		slog.Info("renaming to final output", "from", outputTmp, "to", outputFinal)
		if err := os.Rename(outputTmp, outputFinal); err != nil {
			cleanup(outputTmp)
			cleanup(extractedDB)
			return fmt.Errorf("atomic rename for %s: %w", dayStr, err)
		}

		info, err := os.Stat(outputFinal)
		if err != nil {
			cleanup(extractedDB)
			return fmt.Errorf("stat final output: %w", err)
		}
		slog.Info("day completed", "date", dayStr, "path", outputFinal, "size_bytes", info.Size())

		// Remove extracted DB to reclaim space immediately
		cleanup(extractedDB)
		processed++
	}

	slog.Info("run summary", "processed", processed, "skipped", skipped, "total", len(targetDates))

	return nil
}