bsdaily/internal/bsdaily/run.go

package bsdaily

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

var errEmptySource = errors.New("source file is empty")

// outputDirPerm is world-readable because the dailies tree is
// published for download.
const outputDirPerm = 0o755

// 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) {
	err := os.Remove(path)
	if err != nil && !os.IsNotExist(err) {
		slog.Warn("failed to remove temporary file", "path", path, "error", err)
	}
}

// Run extracts each requested day from the latest daily snapshot into
// a compressed SQL dump. When targetDates is empty it defaults to the
// snapshot date minus one day.
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
	err = CheckFreeSpace(TmpBase, MinTmpFreeBytes, "tmpBase")
	if err != nil {
		return err
	}

	err = CheckFreeSpace(DailiesBase, MinDailiesFreeBytes, "dailiesBase")
	if 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)

		rerr := os.RemoveAll(tmpDir)
		if rerr != nil {
			slog.Error("failed to remove temp directory",
				"path", tmpDir, "error", rerr)
		}
	}()

	// Copy database files from snapshot to temp
	dstDB, err := copySnapshotFiles(snapshotDir, tmpDir)
	if err != nil {
		return 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 {
		didProcess, perr := processDay(tmpDir, dstDB, targetDay)
		if perr != nil {
			return perr
		}

		if didProcess {
			processed++
		} else {
			skipped++
		}
	}

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

	return nil
}

// copySnapshotFiles copies the database, WAL, and (if present) SHM
// files from the snapshot directory into tmpDir and returns the path
// of the copied database.
func copySnapshotFiles(snapshotDir, tmpDir string) (string, error) {
	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("%w: %s", errEmptySource, f)
		}

		slog.Info("source file", "path", f, "size_bytes", info.Size())
	}

	err := CopyFile(srcDB, dstDB)
	if err != nil {
		return "", fmt.Errorf("copying database: %w", err)
	}

	err = CopyFile(srcWAL, dstWAL)
	if err != nil {
		return "", fmt.Errorf("copying WAL: %w", err)
	}

	_, err = os.Stat(srcSHM)
	if err == nil {
		err = CopyFile(srcSHM, dstSHM)
		if err != nil {
			return "", fmt.Errorf("copying SHM: %w", err)
		}
	}

	return dstDB, nil
}

// processDay extracts, dumps, compresses, verifies, and publishes a
// single day. It reports whether the day was processed; false means it
// was skipped (already present or no posts).
func processDay(tmpDir, dstDB string, targetDay time.Time) (bool, error) {
	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")

	_, err := os.Stat(outputFinal)
	if err == nil {
		slog.Info("output already exists, skipping", "path", outputFinal)

		return false, nil
	}

	// Extract target day into a per-day database
	extractedDB := filepath.Join(tmpDir, "extracted-"+dayStr+".db")

	slog.Info("extracting target day", "src", dstDB, "dst", extractedDB)

	err = ExtractDay(dstDB, extractedDB, targetDay)
	if err != nil {
		if errors.Is(err, ErrNoPosts) {
			slog.Warn("no posts found, skipping day", "date", dayStr)
			cleanup(extractedDB)

			return false, nil
		}

		return false, fmt.Errorf("extracting day %s: %w", dayStr, err)
	}

	// Dump to SQL and compress
	err = os.MkdirAll(outputDir, outputDirPerm)
	if err != nil {
		cleanup(extractedDB)

		return false, 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)

	err = DumpAndCompress(extractedDB, outputTmp)
	if err != nil {
		cleanup(outputTmp)
		cleanup(extractedDB)

		return false, fmt.Errorf("dump and compress for %s: %w", dayStr, err)
	}

	slog.Info("verifying compressed output")

	err = VerifyOutput(outputTmp)
	if err != nil {
		cleanup(outputTmp)
		cleanup(extractedDB)

		return false, fmt.Errorf("verification failed for %s: %w", dayStr, err)
	}

	err = publishOutput(outputTmp, outputFinal, dayStr)
	if err != nil {
		cleanup(extractedDB)

		return false, err
	}

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

	return true, nil
}

// publishOutput atomically renames the temporary output file to its
// final path and logs the completed day.
func publishOutput(outputTmp, outputFinal, dayStr string) error {
	// Atomic rename to final path
	slog.Info("renaming to final output", "from", outputTmp, "to", outputFinal)

	err := os.Rename(outputTmp, outputFinal)
	if err != nil {
		cleanup(outputTmp)

		return fmt.Errorf("atomic rename for %s: %w", dayStr, err)
	}

	info, err := os.Stat(outputFinal)
	if err != nil {
		return fmt.Errorf("stat final output: %w", err)
	}

	slog.Info("day completed", "date", dayStr,
		"path", outputFinal, "size_bytes", info.Size())

	return nil
}