ultisuite-backend/internal/migration/admin_jobs.go

package migration

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"

	"github.com/jackc/pgx/v5"
)

// AdminJob is a migration job enriched with the invited user email for admin dashboards.
type AdminJob struct {
	Job
	UserEmail string `json:"user_email"`
}

func (s *Service) ListProjectJobs(ctx context.Context, projectID string) ([]AdminJob, error) {
	rows, err := s.db.Query(ctx, `
		SELECT j.id::text, j.project_id::text, j.user_id::text, j.service, j.status,
		       j.cursor_json, j.stats_json, j.error, j.started_at::text, j.completed_at::text,
		       COALESCE(i.email, '')
		FROM migration_jobs j
		LEFT JOIN migration_invites i
		  ON i.project_id = j.project_id AND i.user_id = j.user_id
		WHERE j.project_id = $1::uuid
		ORDER BY COALESCE(i.email, ''), j.service ASC
	`, projectID)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	var out []AdminJob
	for rows.Next() {
		var row AdminJob
		var cursorRaw, statsRaw []byte
		if err := rows.Scan(
			&row.ID, &row.ProjectID, &row.UserID, &row.Service, &row.Status,
			&cursorRaw, &statsRaw, &row.Error, &row.StartedAt, &row.CompletedAt,
			&row.UserEmail,
		); err != nil {
			return nil, err
		}
		_ = json.Unmarshal(cursorRaw, &row.CursorJSON)
		_ = json.Unmarshal(statsRaw, &row.StatsJSON)
		if row.CursorJSON == nil {
			row.CursorJSON = map[string]any{}
		}
		if row.StatsJSON == nil {
			row.StatsJSON = map[string]any{}
		}
		out = append(out, row)
	}
	return out, rows.Err()
}

func (s *Service) RetryJob(ctx context.Context, projectID, jobID string) (Job, error) {
	var row Job
	var cursorRaw, statsRaw []byte
	err := s.db.QueryRow(ctx, `
		UPDATE migration_jobs
		SET status = 'pending', error = '', updated_at = NOW()
		WHERE id = $1::uuid AND project_id = $2::uuid AND status = 'failed'
		RETURNING id::text, project_id::text, user_id::text, service, status,
		          cursor_json, stats_json, error, started_at::text, completed_at::text
	`, jobID, projectID).Scan(
		&row.ID, &row.ProjectID, &row.UserID, &row.Service, &row.Status,
		&cursorRaw, &statsRaw, &row.Error, &row.StartedAt, &row.CompletedAt,
	)
	if errors.Is(err, pgx.ErrNoRows) {
		return Job{}, fmt.Errorf("job not found or not retryable")
	}
	if err != nil {
		return Job{}, err
	}
	_ = json.Unmarshal(cursorRaw, &row.CursorJSON)
	_ = json.Unmarshal(statsRaw, &row.StatsJSON)
	if row.CursorJSON == nil {
		row.CursorJSON = map[string]any{}
	}
	if row.StatsJSON == nil {
		row.StatsJSON = map[string]any{}
	}
	return row, nil
}

func (s *Service) RetryFailedJobs(ctx context.Context, projectID string) (int64, error) {
	tag, err := s.db.Exec(ctx, `
		UPDATE migration_jobs
		SET status = 'pending', error = '', updated_at = NOW()
		WHERE project_id = $1::uuid AND status = 'failed'
	`, projectID)
	if err != nil {
		return 0, err
	}
	return tag.RowsAffected(), nil
}

func (s *Service) ResetJobCursor(ctx context.Context, projectID, jobID string) (Job, error) {
	tx, err := s.db.Begin(ctx)
	if err != nil {
		return Job{}, err
	}
	defer tx.Rollback(ctx)

	var status string
	err = tx.QueryRow(ctx, `
		SELECT status FROM migration_jobs
		WHERE id = $1::uuid AND project_id = $2::uuid
	`, jobID, projectID).Scan(&status)
	if errors.Is(err, pgx.ErrNoRows) {
		return Job{}, fmt.Errorf("job not found")
	}
	if err != nil {
		return Job{}, err
	}
	if status == "running" {
		return Job{}, fmt.Errorf("job running; wait for completion before reset")
	}

	if _, err := tx.Exec(ctx, `
		DELETE FROM migration_imported_items WHERE job_id = $1::uuid
	`, jobID); err != nil {
		return Job{}, err
	}

	var row Job
	var cursorRaw, statsRaw []byte
	err = tx.QueryRow(ctx, `
		UPDATE migration_jobs
		SET status = 'pending',
		    cursor_json = '{}'::jsonb,
		    stats_json = '{}'::jsonb,
		    error = '',
		    started_at = NULL,
		    completed_at = NULL,
		    updated_at = NOW()
		WHERE id = $1::uuid AND project_id = $2::uuid
		RETURNING id::text, project_id::text, user_id::text, service, status,
		          cursor_json, stats_json, error, started_at::text, completed_at::text
	`, jobID, projectID).Scan(
		&row.ID, &row.ProjectID, &row.UserID, &row.Service, &row.Status,
		&cursorRaw, &statsRaw, &row.Error, &row.StartedAt, &row.CompletedAt,
	)
	if err != nil {
		return Job{}, err
	}
	if err := tx.Commit(ctx); err != nil {
		return Job{}, err
	}
	_ = json.Unmarshal(cursorRaw, &row.CursorJSON)
	_ = json.Unmarshal(statsRaw, &row.StatsJSON)
	if row.CursorJSON == nil {
		row.CursorJSON = map[string]any{}
	}
	if row.StatsJSON == nil {
		row.StatsJSON = map[string]any{}
	}
	return row, nil
}