thanks/internal/jobs/jobs.go

// Package jobs provides definitions and features for Backup Jobs themselves.
// [BackupJob] is the primary resource here.
// [BackupJob.Do] is the main entrypoint for running through the backup process.
package jobs

import (
	"context"
	"fmt"
	"log"
	"os"
	"os/exec"
	"strings"
	"time"

	"git.gentoo.party/sam/thanks/internal/executor"
	"git.gentoo.party/sam/thanks/internal/zfs"
)

type BackupHookCommand struct {
	Command string   `yaml:"command"`
	Args    []string `yaml:"args"`
	Env     []string `yaml:"env"`
}

type BackupHooks struct {
	Completed BackupHookCommand `yaml:"completed"` // called upon successful completion of a backup job
	Error     BackupHookCommand `yaml:"error"`
}

type BackupJob struct {
	Source     string        `yaml:"source"`     // the source dataset (e.g., zroot)
	TargetHost string        `yaml:"targetHost"` // SSH-compatible host
	Target     string        `yaml:"target"`     // the target dataset
	Keep       int           `yaml:"keep"`       // number of snapshots to keep
	MaxAge     time.Duration `yaml:"maxAge"`     // age at which to delete snapshot (if Keep is met)
	Recursive  bool          `yaml:"recursive"`  // create recursive snapshots
	Prefix     string        `yaml:"prefix"`     // name each snapshot with this prefix
	Hooks      BackupHooks   `yaml:"hooks"`      // external programs (libnotify, sendmail, etc) to call

	localExecutor  executor.Executor
	remoteExecutor executor.Executor
}

func (j *BackupJob) listSnapshots(ctx context.Context) ([]zfs.Snapshot, error) {
	allSnaps, err := zfs.Snapshots(ctx, j.localExecutor, j.Source)
	if err != nil {
		return nil, err
	}

	filtered := make([]zfs.Snapshot, 0)
	for _, snap := range allSnaps {
		if !strings.Contains(snap.Name, j.Prefix) {
			continue
		}
		filtered = append(filtered, snap)
	}

	return filtered, nil
}

func (j *BackupJob) snapName() string {
	ts := time.Now().UnixMicro()
	if j.Prefix == "" {
		return fmt.Sprintf("%d", ts)
	} else {
		return fmt.Sprintf("%s%d", j.Prefix, ts)
	}
}

func (j *BackupJob) Snapshot(ctx context.Context) (string, error) {
	snapName := j.snapName()
	snap := fmt.Sprintf("%s@%s", j.Source, snapName)
	err := zfs.CreateSnapshot(
		ctx,
		j.localExecutor,
		j.Source,
		snapName,
	)
	return snap, err
}

func (j *BackupJob) FullSend(ctx context.Context, snap string) error {
	out, err := zfs.FullSend(
		ctx,
		snap,
		j.TargetHost,
		j.Target,
	)
	if err != nil {
		return fmt.Errorf("zfs-send: error: %s - %s", err.Error(), out)
	}
	return nil
}

func (j *BackupJob) IncrementalSend(ctx context.Context, prevSnap *zfs.Snapshot, newSnap string) ([]byte, error) {
	return zfs.IncrementalSend(
		ctx,
		prevSnap.SnapshotName,
		newSnap,
		j.TargetHost,
		j.Target,
	)
}

func (j *BackupJob) Retain(ctx context.Context) error {
	snaps, err := j.listSnapshots(ctx)
	if err != nil {
		return fmt.Errorf("retain: failure listing snapshots: %s", err.Error())
	}

	forDeletion := make([]zfs.Snapshot, 0, len(snaps))
	now := time.Now()

	if len(snaps) < j.Keep {
		return nil
	}

	deleteCount := len(snaps) - j.Keep

	for i, snap := range snaps {
		age := now.Sub(snap.Creation)
		if age >= j.MaxAge {
			forDeletion[i] = snap
		}

		if len(forDeletion) == deleteCount {
			break
		}
	}

	return nil
}

func (j *BackupJob) findCommonAnscestor(ctx context.Context) {
	localSnapshots, err := j.listSnapshots(ctx)
}

type HookErr struct {
	message string
}

func (e *HookErr) Error() string {
	return fmt.Sprintf("error while running hook: %s", e.message)
}

func (j *BackupJob) Do(ctx context.Context) error {
	err := j.do(ctx)
	if err != nil {
		log.Print("job encountered error, running error hook\n")
		err = j.runHookError(ctx, err)
		if err != nil {
			log.Printf("error running hook! %s", err.Error())
		}
		return &HookErr{message: err.Error()}

	} else {
		err = j.runHookCompleted(ctx)
		return &HookErr{message: err.Error()}
	}

	return err
}

func (j *BackupJob) do(ctx context.Context) error {
	currentSnaps, err := j.listSnapshots(ctx)
	if err != nil {
		return fmt.Errorf("failed to list existing snapshots: %s", err.Error())
	}
	var prev *zfs.Snapshot

	if len(currentSnaps) > 0 {
		prev = &currentSnaps[len(currentSnaps)-1]
	}

	newSnapshot, err := j.Snapshot(ctx)
	if err != nil {
		return fmt.Errorf("error creating new snapshot: %s", err.Error())
	}

	var sendErr error
	if prev == nil {
		// Full Send
		sendErr = j.FullSend(ctx, newSnapshot)
	} else {
		// Inc send
		_, sendErr = j.IncrementalSend(ctx, prev, newSnapshot)
	}

	if sendErr != nil {
		return fmt.Errorf("error sending backup: %s", sendErr.Error())
	}

	return nil
}

func (j *BackupJob) runHookCompleted(ctx context.Context) error {
	cmdDef := j.Hooks.Completed
	if cmdDef.Command == "" {
		return nil
	}

	timeoutCtx, cancel := context.WithTimeout(
		ctx,
		time.Second*3,
	)
	cmd := exec.CommandContext(timeoutCtx, cmdDef.Command, cmdDef.Args...)
	defer cancel()
	cmd.Env = []string{
		fmt.Sprintf("THANKS_JOB_SOURCE=%s", j.Source),
		fmt.Sprintf("THANKS_JOB_TARGET=%s", j.Target),
	}
	return cmd.Run()
}

func (j *BackupJob) runHookError(ctx context.Context, err error) error {
	cmdDef := j.Hooks.Error
	if cmdDef.Command == "" {
		return nil
	}

	timeoutCtx, cancel := context.WithTimeout(
		ctx,
		time.Second*3,
	)
	cmd := exec.CommandContext(timeoutCtx, cmdDef.Command, cmdDef.Args...)
	defer cancel()
	cmd.Env = append(
		[]string{
			fmt.Sprintf("THANKS_JOB_SOURCE=%s", j.Source),
			fmt.Sprintf("THANKS_JOB_TARGET=%s", j.Target),
			fmt.Sprintf("THANKS_JOB_ERROR=%s", err.Error()),
			fmt.Sprintf("DISPLAY=%s", os.Getenv("DISPLAY")),
		},
		cmdDef.Env...,
	)

	out, runErr := cmd.CombinedOutput()
	if runErr != nil {
		log.Printf("%s", out)
	}

	return runErr
}