package migrate // import "git.griefed.de/griefed/repoman/internal/migrate" FUNCTIONS func BuildArgs( cfg *migrationconfig.MigrationConfig, gs *globalconfig.Settings, ownerTokens map[string]string, mode run.RunMode, ) []string BuildArgs converts a MigrationConfig and the active global settings into the argument slice for migrate.sh. The script is always called with --no-env so it reads nothing from the process environment. All configuration is explicit via --set. --force-sync is always passed to suppress interactive prompts. The resulting slice is suitable for exec.Command("migrate.sh", args...). func RunLogBase(configName string) string RunLogBase is the RotatingWriter base name for a config's per-run log files, so the writer here and any reader (the /configs log download) derive the same names from a config name instead of duplicating the sanitising rules and drifting. The files on disk are "<base>-YYYY-MM-DD.log" under the worker's RunLogDir. TYPES type EventKind uint8 EventKind classifies a single line of migrate.sh output. const ( KindLog EventKind = iota // generic stdout line KindOK // [OK] prefix KindSkip // [SKIP] prefix KindError // [ERROR] prefix KindDryRun // [DRY-RUN] prefix KindProgress // \r in-place progress bar — not stored in DB KindSummary // key: value summary block KindFatal // pre-flight failure (script exits 1) KindWarn // "⚠" line: something did not work, run continues ) type FakeRunner struct { Lines []string ExitCode int Delay time.Duration // optional delay between lines } FakeRunner emits a pre-canned sequence of lines and an exit code. Use it in tests that exercise the worker or handler without a real script. func (f *FakeRunner) Preflight(_ *WorkDir) error Preflight is a no-op for the fake runner — it needs no external script. func (f *FakeRunner) Start(_ context.Context, _ RunConfig) (*RunHandle, error) Start emits the fake's canned lines (with optional inter-line delay) and finishes with its configured exit code, ignoring the context and config. type GoRepoRecord struct { ForgejoOwner string ForgejoName string OwnerType string OwningConfigID *int64 SourcePlatform string SourceNamespace string SourceOwnerType string IsPrivate bool IsArchived bool } GoRepoRecord is the data the GoRunner passes to RepoUpsert after each successful migration. The worker wires this to db.RepositoryRepo.UpsertRepo. type GoRunner struct { // MigrateTimeout caps a single (synchronous) Forgejo migration. Zero falls // back to forgejo.DefaultMigrateTimeout. Wired from the --migrate-timeout flag. MigrateTimeout time.Duration } GoRunner is the Phase-5 native Go Runner. It replaces ScriptRunner: 1. For each Source in the config: a. Build a source.Client for that platform. b. Fetch the full repo list from the source API. c. For each repo: - Check if it already exists in Forgejo → [SKIP]. - If dry-run → [DRY-RUN] and continue. - Call Forgejo migration API → [OK] or [ERROR]. - Upsert into the repoman repositories table. 2. For outbound targets (push mirrors), add each enabled target as a push mirror on the Forgejo repo after import. GoRunner implements the same Runner interface as ScriptRunner, so the worker, SSE handler, and all other callers are completely unchanged. func NewGoRunner() *GoRunner NewGoRunner returns a GoRunner ready to use. func (gr *GoRunner) Preflight(_ *WorkDir) error Preflight is a no-op: the native engine needs no migrate.sh on disk, so a GoRunner run can always be queued. func (gr *GoRunner) Start(ctx context.Context, cfg RunConfig) (*RunHandle, error) Start launches the migration in a goroutine and returns a RunHandle. type IssueImportTarget struct { ConfigID int64 ForgejoOwner string ForgejoRepo string SourcePlatform string SourceNamespace string SourceRepo string } IssueImportTarget identifies the one Forgejo repo and the source repo whose issues are imported into it — the migrate-local mirror of issuesync.RepoLink, kept here so RunConfig.ImportIssues carries no issuesync dependency. ConfigID lets the import callback address the per-config webhook secret + ingress URL when it provisions the Forgejo hook. type ParsedLine struct { Kind EventKind Level run.EventLevel Source string Message string // Repo is non-empty for OK/Skip/Error/DryRun kinds. Repo string } ParsedLine is the result of parsing one output line from the script. func ParseLine(raw string) ParsedLine ParseLine classifies and normalises one raw output line. The caller is responsible for splitting the raw combined stdout/stderr stream on newlines. Lines that consist solely of whitespace after stripping are returned as KindLog with an empty Message. func (p ParsedLine) ToEvent(runID int64, at time.Time) run.Event ToEvent converts a ParsedLine into a run.Event ready for DB storage. KindProgress events should NOT be stored; callers must check Kind first. The caller provides the runID and timestamp. type Result struct { ExitCode int Counters run.Counters // Err is non-nil only for infrastructure failures (fork/exec, // context cancellation) — not for script-level errors, which // are reported via exit code. Err error } Result is the final outcome of a run. func (r Result) ExitStatus() run.ExitStatus ExitStatus maps a Result to the run.ExitStatus enum. type RunConfig struct { // RunID is the DB row ID of the run already created by the worker. RunID int64 // ConfigID is used to locate the working directory. ConfigID int64 // Mode is the per-run mode from the queue entry (real / dry-run / // first-run-forced-dry). GoRunner reads it directly to decide whether to // write; ScriptRunner receives the same information baked into Args by // BuildArgs. Both route the decision through isDryRun so they agree. Mode run.RunMode // ── ScriptRunner fields (ignored by GoRunner) ────────────── WorkDir string ScriptPath string Args []string // ── GoRunner fields (ignored by ScriptRunner) ────────────── // MigrationConfig and GlobalSettings are populated by the worker // when using GoRunner. ScriptRunner uses Args instead. MigrationConfig *migrationconfigpkg.MigrationConfig GlobalSettings *globalconfigpkg.Settings // OwnerTokens is the config owner's account source tokens, keyed by // migrationconfig.OwnerTokenKey(platform, instanceURL) → token — the reference // model's primary source of token VALUES and the fallback in token resolution // (per-source → config override → owner token). The worker loads it from the // config owner's user_platform_tokens; nil when the config has no owner or no // tokens. OwnerTokens map[string]string RepoUpsert func(ctx context.Context, r GoRepoRecord) error // ImportIssues, when non-nil AND the config's Settings.SyncIssues is on, // imports a source repo's issues + comments into the Forgejo hub (the // issue-hub feature). The worker wires it to issuesync.Importer; it stays a // callback — passed the run's own forgejo client and a source.IssueClient — // so the engine never imports the issuesync/db layer (the same decoupling // seam as RepoUpsert). Invoked only on non-dry runs. ImportIssues func(ctx context.Context, fc *forgejo.Client, src source.IssueClient, target IssueImportTarget, emit func(string)) error } RunConfig is the complete specification for one run. type RunHandle struct { // Events streams parsed output lines. The consumer should read // from this channel until it is closed. Events <-chan ParsedLine // Done carries the final result. It receives exactly one value, // then is closed. Done <-chan Result } RunHandle is returned by Runner.Start. Both channels are closed when the run finishes. type Runner interface { // Start begins the migration and returns a RunHandle through // which the caller receives events and the final result. // The caller MUST drain Events and wait on Done. Start(ctx context.Context, cfg RunConfig) (*RunHandle, error) // Preflight validates that this runner can execute a run, given the // working-directory manager, and returns an error the caller can surface // BEFORE a run is queued. GoRunner needs nothing (returns nil); ScriptRunner // checks migrate.sh is present and executable. The worker calls this from // Enqueue so a misconfiguration is reported at trigger time rather than // failing the run at execution time. Preflight(workDir *WorkDir) error } Runner is the interface for starting a migration run. The real implementation execs migrate.sh; tests inject a fake. type ScriptRunner struct { // CmdFactory builds the exec.Cmd to run. Injectable for tests. // If nil, defaults to exec.CommandContext. CmdFactory func(ctx context.Context, name string, args ...string) *exec.Cmd } ScriptRunner is the production Runner. It execs migrate.sh as a subprocess, reads its combined stdout+stderr line by line, and emits ParsedLine events. func NewScriptRunner() *ScriptRunner NewScriptRunner returns a production ScriptRunner. func (sr *ScriptRunner) Preflight(workDir *WorkDir) error Preflight checks migrate.sh is present and executable, so a missing or non-executable script is reported when a run is queued rather than failing the run at exec time. func (sr *ScriptRunner) Start(ctx context.Context, cfg RunConfig) (*RunHandle, error) Start launches migrate.sh and returns a RunHandle. stdout and stderr are merged into one stream using cmd.StderrPipe piped through the same io.Reader. type UserTokenLoader interface { TokensFor(ctx context.Context, userID int64) (map[string]string, error) } UserTokenLoader loads a user's personal source tokens (platform→token), used as the per-owner fallback in token resolution. db.UserPlatformTokenRepo satisfies it. type WorkDir struct { // Has unexported fields. } WorkDir manages the per-config working directory used by migrate.sh. Each config gets its own directory so run state files — particularly resume-state.log and .first-run-completed — are isolated and survive across runs of the same config. Layout: {dataDir}/migrate/ {configID}/ migration-logs/ ← created by migrate.sh success_*.log error_*.log summary_*.log resume-state.log ← cumulative; survives runs .first-run-completed ← sentinel; survives runs func NewWorkDir(dataDir string) *WorkDir NewWorkDir constructs a WorkDir rooted at dataDir. func (w *WorkDir) For(configID int64) (string, error) For returns the working directory path for the given config ID, creating it (and its migration-logs subdirectory) if necessary. func (w *WorkDir) ScriptPath() string ScriptPath returns the path to the migrate.sh script. The script must be present at {dataDir}/migrate.sh. Callers should verify this with Validate before starting a run. func (w *WorkDir) Validate() error Validate checks that migrate.sh is present and executable. type Worker struct { Runs run.Repository Events run.EventRepository Queue run.QueueRepository Configs migrationconfig.Repository GlobalSvc *globalconfig.Service Runner Runner WorkDir *WorkDir Now func() time.Time // PollInterval is how often the queue is polled for new work. Defaults to // 2s when unset; tests set it small for fast, deterministic execution. PollInterval time.Duration // RunLogDir, when set, is the directory for per-config run logs: each run // (dry or real) is written to "run-<config>-YYYY-MM-DD.log" there, with the // same date-rotation as the app/security logs. Empty disables run-file // logging (events still go to the DB and SSE). RunLogDir string RunLogRetentionDays int // RepoUpsertFn is called after each successful repo migration to // record the repo in the inventory. If nil, inventory updates are // skipped (acceptable in tests that don't exercise the DB layer). RepoUpsertFn func(ctx context.Context, record GoRepoRecord) error // IssueImportFn, when set, imports a repo's source issues + comments into the // Forgejo hub during a run (the issue-hub feature). serve.go wires it to an // issuesync.Importer; nil disables issue sync regardless of a config's // SyncIssues setting (acceptable in tests). It is passed the run's own forgejo // client so it imports against the same instance the run writes to. IssueImportFn func(ctx context.Context, fc *forgejo.Client, src source.IssueClient, target IssueImportTarget, emit func(string)) error // UserTokens loads a config owner's personal source tokens (the multi-tenant // fallback resolved after per-source/per-config tokens). Optional; nil means a // config falls back to no owner token (only per-source/per-config tokens work). UserTokens UserTokenLoader // Audit records run lifecycle events (triggered, completed). Optional; // a nil Logger is a no-op, so tests may leave it unset. Audit *audit.Logger // Has unexported fields. } Worker is the run-queue FIFO worker. It: - Polls the run queue on a ticker (default 2 s). - Starts at most one run at a time. - Writes events to the EventRepository as they arrive. - Updates the Run row on completion. - Broadcasts events to any registered SSE subscribers. Construct with NewWorker; start with Worker.Run(ctx). func NewWorker( runs run.Repository, events run.EventRepository, queue run.QueueRepository, configs migrationconfig.Repository, globalSvc *globalconfig.Service, runner Runner, workDir *WorkDir, ) *Worker NewWorker constructs a Worker with default clock. func (w *Worker) ActiveRunID() int64 ActiveRunID returns the ID of the currently-executing run, or 0. func (w *Worker) Cancel(runID int64) bool Cancel requests cancellation of runID when it is the active run, returning true if a cancellation was signalled. The runner honours context cancellation, so the run stops promptly and execute records it as cancelled (ExitCode 130 → StatusCancelled) as it unwinds. Returns false when runID is not the active run (already finished or never started), so the caller can report "not active". func (w *Worker) Enqueue(ctx context.Context, entry *run.QueueEntry) error Enqueue adds a run to the queue. Deduplication is on by default for scheduled triggers; manual triggers always queue (dedup=false). It first runs the active runner's Preflight so a misconfiguration (e.g. a missing migrate.sh in --script-runner mode) is rejected here — where the trigger handler surfaces it as a flash — instead of accepting the job and failing it at execution time. func (w *Worker) RecoverOrphans(ctx context.Context) error RecoverOrphans reconciles runs left in StatusRunning by a previous process and must be called ONCE at startup, before Run. A daemon restart/crash orphans the in-flight run: its queue entry was already popped (so it never re-runs) and the finalizing write never landed (the shutdown cancels the context the write would use), so the row is stuck at "running" forever — which also hides the monitor's cancel button and leaves a perpetual "live" badge. Because the worker is single-instance and not yet started, any StatusRunning row here is necessarily such an orphan. Each orphan is recorded as interrupted (a final event + StatusFailed). An orphan whose config is still runnable is then re-queued as a one-shot continuation (IsRecovery=true): the engine is idempotent (skips already-migrated repos, resumes release/issue sync, repairs partial assets), so the continuation picks up where the interrupted run stopped. The loop guard is progress-aware. A continuation that was ITSELF interrupted is resumed again only if it actually advanced the migration before dying; one that crashed without progress is treated as a poison run and stopped, so a config that crashes the daemon on every attempt cannot loop — while a long migration interrupted by an unrelated restart, having made progress, keeps resuming. Every non-continuation outcome is recorded on the orphan's log (so an operator can see WHY a run did or did not resume). Best-effort: a per-orphan failure is logged and the rest still reconcile. Each orphan is assessed (read-only) BEFORE it is finalized, so the progress check reads the run's genuine event history without the interruption marker finalize appends; the human-readable decision is then recorded after, keeping the log in chronological order (interrupted → resumed / not resumed). func (w *Worker) Run(ctx context.Context) Run starts the worker loop. It blocks until ctx is cancelled. func (w *Worker) Subscribe(runID int64) chan run.Event Subscribe registers a channel that will receive events for runID. The channel is buffered; the caller should drain it promptly, and call Unsubscribe with the returned channel when done. If runID is not the currently-active run (it already finished, or never started), Subscribe returns an already-CLOSED channel. Only the active run's completion closes per-run subscriber channels, so a channel registered for a non-active run would never close and the subscriber would block forever — the race a caller hits when a run finishes between an IsFinished() check and this call. The active check and registration are atomic with completion (same mutex as execute's finish defer), so a channel is either guaranteed to be closed by that run's completion or returned already closed. func (w *Worker) Unsubscribe(runID int64, ch chan run.Event) Unsubscribe removes a subscription and closes the channel.