570 lines
15 KiB
Go
570 lines
15 KiB
Go
|
package main
|
||
|
|
|
||
|
|
import (
|
||
|
|
"bufio"
|
||
|
|
"context"
|
||
|
|
"fmt"
|
||
|
|
"io"
|
||
|
|
"log"
|
||
|
|
"os"
|
||
|
|
"strconv"
|
||
|
|
"strings"
|
||
|
|
"time"
|
||
|
|
|
||
|
|
"github.com/descrybe/descrybe-v2/apps/api/internal/billing"
|
||
|
|
"github.com/google/uuid"
|
||
|
|
"github.com/jackc/pgx/v5/pgxpool"
|
||
|
|
)
|
||
|
|
|
||
|
|
// recoverJobsFromMySQLDump streams a mysqldump and upserts every A1
|
||
|
|
// processing_jobs (+ joinable processing_job_products) into live Postgres.
|
||
|
|
//
|
||
|
|
// Needed because live A1 often only retains a couple of recent jobs (retention
|
||
|
|
// deletes non-migrated terminal rows after 30 days; jobs domain may never have
|
||
|
|
// been imported). Export alone cannot invent history that is missing from PG.
|
||
|
|
//
|
||
|
|
// Job products resolve raw_product_id via dump GTIN → Postgres raw_products.
|
||
|
|
func recoverJobsFromMySQLDump(ctx context.Context, pg *pgxpool.Pool, companyID uuid.UUID, dumpPath string) error {
|
||
|
|
legacyCompany := billing.A1LegacyCompanyID
|
||
|
|
var legacy string
|
||
|
|
_ = pg.QueryRow(ctx, `SELECT COALESCE(legacy_company_id::text, '') FROM companies WHERE id = $1`, companyID).Scan(&legacy)
|
||
|
|
if legacy != "" {
|
||
|
|
legacyCompany = legacy
|
||
|
|
}
|
||
|
|
|
||
|
|
fi, err := os.Stat(dumpPath)
|
||
|
|
if err != nil {
|
||
|
|
return fmt.Errorf("stat mysql dump %q: %w", dumpPath, err)
|
||
|
|
}
|
||
|
|
log.Printf("recover-jobs: dump=%s size=%d legacy_company=%s", dumpPath, fi.Size(), legacyCompany)
|
||
|
|
|
||
|
|
userByLegacy, err := loadUserLegacyMap(ctx, pg, companyID)
|
||
|
|
if err != nil {
|
||
|
|
return err
|
||
|
|
}
|
||
|
|
|
||
|
|
f, err := os.Open(dumpPath)
|
||
|
|
if err != nil {
|
||
|
|
return fmt.Errorf("open mysql dump: %w", err)
|
||
|
|
}
|
||
|
|
defer f.Close()
|
||
|
|
|
||
|
|
jobs, err := scanA1ProcessingJobs(f, legacyCompany)
|
||
|
|
if err != nil {
|
||
|
|
return err
|
||
|
|
}
|
||
|
|
if len(jobs) == 0 {
|
||
|
|
return fmt.Errorf("no processing_jobs for legacy company %s in dump", legacyCompany)
|
||
|
|
}
|
||
|
|
log.Printf("dump: %d A1 processing_jobs (incl. history)", len(jobs))
|
||
|
|
|
||
|
|
if _, err := f.Seek(0, io.SeekStart); err != nil {
|
||
|
|
return err
|
||
|
|
}
|
||
|
|
jobIDs := make(map[string]struct{}, len(jobs))
|
||
|
|
for _, j := range jobs {
|
||
|
|
jobIDs[j.id] = struct{}{}
|
||
|
|
}
|
||
|
|
pjpRows, rawLegacyIDs, err := scanA1JobProducts(f, jobIDs)
|
||
|
|
if err != nil {
|
||
|
|
return err
|
||
|
|
}
|
||
|
|
log.Printf("dump: %d A1 processing_job_products across %d raw ids", len(pjpRows), len(rawLegacyIDs))
|
||
|
|
|
||
|
|
if _, err := f.Seek(0, io.SeekStart); err != nil {
|
||
|
|
return err
|
||
|
|
}
|
||
|
|
gtinByLegacy, err := scanRawProductGTINs(f, rawLegacyIDs, legacyCompany)
|
||
|
|
if err != nil {
|
||
|
|
return err
|
||
|
|
}
|
||
|
|
log.Printf("dump: resolved %d/%d raw→gtin mappings", len(gtinByLegacy), len(rawLegacyIDs))
|
||
|
|
|
||
|
|
rawByGTIN, err := loadRawByGTIN(ctx, pg, companyID, gtinByLegacy)
|
||
|
|
if err != nil {
|
||
|
|
return err
|
||
|
|
}
|
||
|
|
|
||
|
|
var jobsUpserted, jobsSkipped int
|
||
|
|
for _, j := range jobs {
|
||
|
|
jobUUID, err := parseDumpJobID(j.id)
|
||
|
|
if err != nil {
|
||
|
|
jobsSkipped++
|
||
|
|
continue
|
||
|
|
}
|
||
|
|
var userID *uuid.UUID
|
||
|
|
if uid, ok := userByLegacy[j.userLegacy]; ok {
|
||
|
|
userID = &uid
|
||
|
|
}
|
||
|
|
status := normalizeProcessingJobStatus(j.status)
|
||
|
|
ptype := strings.TrimSpace(j.processingType)
|
||
|
|
if ptype == "" {
|
||
|
|
ptype = "full"
|
||
|
|
}
|
||
|
|
var errPtr *string
|
||
|
|
if j.errText != "" {
|
||
|
|
v := j.errText
|
||
|
|
errPtr = &v
|
||
|
|
}
|
||
|
|
_, err = pg.Exec(ctx, `
|
||
|
|
INSERT INTO processing_jobs (
|
||
|
|
id, company_id, user_id, status, total_products, processed_products,
|
||
|
|
error, processing_type, priority, estimated_tokens,
|
||
|
|
started_at, completed_at, created_at, updated_at,
|
||
|
|
current_step, step_progress, ai_provider_mode
|
||
|
|
) VALUES (
|
||
|
|
$1, $2, $3, $4, $5, $6,
|
||
|
|
$7, $8, $9, $10,
|
||
|
|
$11, $12, $13, $14,
|
||
|
|
'', '[]'::jsonb, 'migrated'
|
||
|
|
)
|
||
|
|
ON CONFLICT (id) DO UPDATE SET
|
||
|
|
company_id = EXCLUDED.company_id,
|
||
|
|
user_id = COALESCE(EXCLUDED.user_id, processing_jobs.user_id),
|
||
|
|
status = EXCLUDED.status,
|
||
|
|
total_products = EXCLUDED.total_products,
|
||
|
|
processed_products = EXCLUDED.processed_products,
|
||
|
|
error = EXCLUDED.error,
|
||
|
|
processing_type = EXCLUDED.processing_type,
|
||
|
|
priority = EXCLUDED.priority,
|
||
|
|
estimated_tokens = EXCLUDED.estimated_tokens,
|
||
|
|
started_at = EXCLUDED.started_at,
|
||
|
|
completed_at = EXCLUDED.completed_at,
|
||
|
|
created_at = EXCLUDED.created_at,
|
||
|
|
updated_at = EXCLUDED.updated_at,
|
||
|
|
ai_provider_mode = 'migrated'`,
|
||
|
|
jobUUID, companyID, userID, status,
|
||
|
|
j.totalProducts, j.processedProducts,
|
||
|
|
errPtr, ptype, j.priority, j.estimatedTokens,
|
||
|
|
j.startedAt, j.completedAt, j.createdAt, j.updatedAt,
|
||
|
|
)
|
||
|
|
if err != nil {
|
||
|
|
log.Printf("job %s: %v", j.id, err)
|
||
|
|
jobsSkipped++
|
||
|
|
continue
|
||
|
|
}
|
||
|
|
jobsUpserted++
|
||
|
|
}
|
||
|
|
|
||
|
|
var pjpUpserted, pjpSkipped int
|
||
|
|
for _, row := range pjpRows {
|
||
|
|
jobUUID, err := parseDumpJobID(row.jobID)
|
||
|
|
if err != nil {
|
||
|
|
pjpSkipped++
|
||
|
|
continue
|
||
|
|
}
|
||
|
|
gtin, ok := gtinByLegacy[row.rawLegacy]
|
||
|
|
if !ok || gtin == "" {
|
||
|
|
pjpSkipped++
|
||
|
|
continue
|
||
|
|
}
|
||
|
|
rawUUID, ok := rawByGTIN[gtin]
|
||
|
|
if !ok {
|
||
|
|
pjpSkipped++
|
||
|
|
continue
|
||
|
|
}
|
||
|
|
prodID := uuid.NewSHA1(uuid.NameSpaceOID, []byte("pjp:"+strconv.FormatInt(row.legacyID, 10)))
|
||
|
|
var errPtr *string
|
||
|
|
if row.errText != "" {
|
||
|
|
v := row.errText
|
||
|
|
errPtr = &v
|
||
|
|
}
|
||
|
|
_, err = pg.Exec(ctx, `
|
||
|
|
INSERT INTO processing_job_products (
|
||
|
|
id, job_id, raw_product_id, status, error, processed_product_id, created_at, updated_at
|
||
|
|
) VALUES ($1, $2, $3, $4, $5, NULL, $6, $7)
|
||
|
|
ON CONFLICT (id) DO UPDATE SET
|
||
|
|
status = EXCLUDED.status,
|
||
|
|
error = EXCLUDED.error,
|
||
|
|
raw_product_id = EXCLUDED.raw_product_id,
|
||
|
|
updated_at = EXCLUDED.updated_at`,
|
||
|
|
prodID, jobUUID, rawUUID, normalizeJobProductStatus(row.status), errPtr,
|
||
|
|
row.createdAt, row.updatedAt,
|
||
|
|
)
|
||
|
|
if err != nil {
|
||
|
|
pjpSkipped++
|
||
|
|
continue
|
||
|
|
}
|
||
|
|
pjpUpserted++
|
||
|
|
}
|
||
|
|
|
||
|
|
var liveJobs, livePJP int
|
||
|
|
_ = pg.QueryRow(ctx, `SELECT count(*) FROM processing_jobs WHERE company_id=$1`, companyID).Scan(&liveJobs)
|
||
|
|
_ = pg.QueryRow(ctx, `SELECT count(*) FROM processing_job_products WHERE job_id IN (SELECT id FROM processing_jobs WHERE company_id=$1)`, companyID).Scan(&livePJP)
|
||
|
|
log.Printf("recover-jobs done: jobs_upserted=%d skipped=%d pjp_upserted=%d skipped=%d live_jobs=%d live_job_products=%d",
|
||
|
|
jobsUpserted, jobsSkipped, pjpUpserted, pjpSkipped, liveJobs, livePJP)
|
||
|
|
return nil
|
||
|
|
}
|
||
|
|
|
||
|
|
type dumpJob struct {
|
||
|
|
id string
|
||
|
|
userLegacy string
|
||
|
|
status, processingType, errText string
|
||
|
|
totalProducts, processedProducts int
|
||
|
|
priority, estimatedTokens int
|
||
|
|
startedAt, completedAt *time.Time
|
||
|
|
createdAt, updatedAt time.Time
|
||
|
|
}
|
||
|
|
|
||
|
|
type dumpJobProduct struct {
|
||
|
|
legacyID int64
|
||
|
|
jobID, rawLegacy string
|
||
|
|
status, errText string
|
||
|
|
createdAt, updatedAt time.Time
|
||
|
|
}
|
||
|
|
|
||
|
|
func loadUserLegacyMap(ctx context.Context, pg *pgxpool.Pool, companyID uuid.UUID) (map[string]uuid.UUID, error) {
|
||
|
|
rows, err := pg.Query(ctx, `
|
||
|
|
SELECT u.id, COALESCE(u.legacy_user_id, '')
|
||
|
|
FROM users u
|
||
|
|
JOIN memberships m ON m.user_id = u.id
|
||
|
|
WHERE m.company_id = $1 AND COALESCE(u.legacy_user_id, '') <> ''`, companyID)
|
||
|
|
if err != nil {
|
||
|
|
return nil, err
|
||
|
|
}
|
||
|
|
defer rows.Close()
|
||
|
|
out := map[string]uuid.UUID{}
|
||
|
|
for rows.Next() {
|
||
|
|
var id uuid.UUID
|
||
|
|
var legacy string
|
||
|
|
if err := rows.Scan(&id, &legacy); err != nil {
|
||
|
|
return nil, err
|
||
|
|
}
|
||
|
|
out[legacy] = id
|
||
|
|
}
|
||
|
|
return out, rows.Err()
|
||
|
|
}
|
||
|
|
|
||
|
|
func loadRawByGTIN(ctx context.Context, pg *pgxpool.Pool, companyID uuid.UUID, gtinByLegacy map[string]string) (map[string]uuid.UUID, error) {
|
||
|
|
uniq := make([]string, 0, len(gtinByLegacy))
|
||
|
|
seen := map[string]struct{}{}
|
||
|
|
for _, g := range gtinByLegacy {
|
||
|
|
g = strings.TrimSpace(g)
|
||
|
|
if g == "" {
|
||
|
|
continue
|
||
|
|
}
|
||
|
|
if _, ok := seen[g]; ok {
|
||
|
|
continue
|
||
|
|
}
|
||
|
|
seen[g] = struct{}{}
|
||
|
|
uniq = append(uniq, g)
|
||
|
|
}
|
||
|
|
out := map[string]uuid.UUID{}
|
||
|
|
if len(uniq) == 0 {
|
||
|
|
return out, nil
|
||
|
|
}
|
||
|
|
rows, err := pg.Query(ctx, `
|
||
|
|
SELECT gtin, id FROM raw_products
|
||
|
|
WHERE company_id = $1 AND gtin = ANY($2)`, companyID, uniq)
|
||
|
|
if err != nil {
|
||
|
|
return nil, err
|
||
|
|
}
|
||
|
|
defer rows.Close()
|
||
|
|
for rows.Next() {
|
||
|
|
var gtin string
|
||
|
|
var id uuid.UUID
|
||
|
|
if err := rows.Scan(>in, &id); err != nil {
|
||
|
|
return nil, err
|
||
|
|
}
|
||
|
|
out[gtin] = id
|
||
|
|
}
|
||
|
|
return out, rows.Err()
|
||
|
|
}
|
||
|
|
|
||
|
|
func scanA1ProcessingJobs(r io.Reader, legacyCompany string) ([]dumpJob, error) {
|
||
|
|
br := bufio.NewReaderSize(r, 4<<20)
|
||
|
|
mode := false
|
||
|
|
var out []dumpJob
|
||
|
|
for {
|
||
|
|
line, err := br.ReadString('\n')
|
||
|
|
if strings.HasPrefix(line, "INSERT INTO `processing_jobs`") {
|
||
|
|
mode = true
|
||
|
|
} else if mode && dumpSectionEnded(line, "processing_jobs") {
|
||
|
|
mode = false
|
||
|
|
}
|
||
|
|
if mode && strings.Contains(line, "'"+legacyCompany+"'") {
|
||
|
|
fields := parseMySQLTupleFields(line)
|
||
|
|
// id, user_id, company_id, status, total, processed, error, started, completed, created, updated, type, priority, estimated
|
||
|
|
if len(fields) >= 14 && fields[2] == legacyCompany {
|
||
|
|
j := dumpJob{
|
||
|
|
id: fields[0],
|
||
|
|
userLegacy: nullish(fields[1]),
|
||
|
|
status: fields[3],
|
||
|
|
totalProducts: atoiDefault(fields[4], 0),
|
||
|
|
processedProducts: atoiDefault(fields[5], 0),
|
||
|
|
errText: nullish(fields[6]),
|
||
|
|
startedAt: parseDumpTimePtr(fields[7]),
|
||
|
|
completedAt: parseDumpTimePtr(fields[8]),
|
||
|
|
createdAt: parseDumpTime(fields[9]),
|
||
|
|
updatedAt: parseDumpTime(fields[10]),
|
||
|
|
processingType: nullish(fields[11]),
|
||
|
|
priority: atoiDefault(fields[12], 0),
|
||
|
|
estimatedTokens: atoiDefault(fields[13], 0),
|
||
|
|
}
|
||
|
|
if j.createdAt.IsZero() {
|
||
|
|
j.createdAt = time.Now().UTC()
|
||
|
|
}
|
||
|
|
if j.updatedAt.IsZero() {
|
||
|
|
j.updatedAt = j.createdAt
|
||
|
|
}
|
||
|
|
out = append(out, j)
|
||
|
|
}
|
||
|
|
}
|
||
|
|
if err == io.EOF {
|
||
|
|
break
|
||
|
|
}
|
||
|
|
if err != nil {
|
||
|
|
return nil, err
|
||
|
|
}
|
||
|
|
}
|
||
|
|
return out, nil
|
||
|
|
}
|
||
|
|
|
||
|
|
func scanA1JobProducts(r io.Reader, jobIDs map[string]struct{}) ([]dumpJobProduct, map[string]struct{}, error) {
|
||
|
|
br := bufio.NewReaderSize(r, 4<<20)
|
||
|
|
mode := false
|
||
|
|
var out []dumpJobProduct
|
||
|
|
rawIDs := map[string]struct{}{}
|
||
|
|
for {
|
||
|
|
line, err := br.ReadString('\n')
|
||
|
|
if strings.HasPrefix(line, "INSERT INTO `processing_job_products`") {
|
||
|
|
mode = true
|
||
|
|
} else if mode && dumpSectionEnded(line, "processing_job_products") {
|
||
|
|
mode = false
|
||
|
|
}
|
||
|
|
if mode && looksLikeTupleLine(line) {
|
||
|
|
fields := parseMySQLTupleFields(line)
|
||
|
|
// id, job_id, raw_product_id, status, error, processed_product_id, created, updated
|
||
|
|
if len(fields) >= 8 {
|
||
|
|
if _, ok := jobIDs[fields[1]]; ok {
|
||
|
|
legacyID, _ := strconv.ParseInt(fields[0], 10, 64)
|
||
|
|
row := dumpJobProduct{
|
||
|
|
legacyID: legacyID,
|
||
|
|
jobID: fields[1],
|
||
|
|
rawLegacy: fields[2],
|
||
|
|
status: fields[3],
|
||
|
|
errText: nullish(fields[4]),
|
||
|
|
createdAt: parseDumpTime(fields[6]),
|
||
|
|
updatedAt: parseDumpTime(fields[7]),
|
||
|
|
}
|
||
|
|
if row.createdAt.IsZero() {
|
||
|
|
row.createdAt = time.Now().UTC()
|
||
|
|
}
|
||
|
|
if row.updatedAt.IsZero() {
|
||
|
|
row.updatedAt = row.createdAt
|
||
|
|
}
|
||
|
|
out = append(out, row)
|
||
|
|
rawIDs[row.rawLegacy] = struct{}{}
|
||
|
|
}
|
||
|
|
}
|
||
|
|
}
|
||
|
|
if err == io.EOF {
|
||
|
|
break
|
||
|
|
}
|
||
|
|
if err != nil {
|
||
|
|
return nil, nil, err
|
||
|
|
}
|
||
|
|
}
|
||
|
|
return out, rawIDs, nil
|
||
|
|
}
|
||
|
|
|
||
|
|
func scanRawProductGTINs(r io.Reader, want map[string]struct{}, legacyCompany string) (map[string]string, error) {
|
||
|
|
br := bufio.NewReaderSize(r, 4<<20)
|
||
|
|
mode := false
|
||
|
|
out := map[string]string{}
|
||
|
|
remaining := len(want)
|
||
|
|
for remaining > 0 {
|
||
|
|
line, err := br.ReadString('\n')
|
||
|
|
if strings.HasPrefix(line, "INSERT INTO `raw_products`") {
|
||
|
|
mode = true
|
||
|
|
} else if mode && dumpSectionEnded(line, "raw_products") {
|
||
|
|
mode = false
|
||
|
|
}
|
||
|
|
if mode && looksLikeTupleLine(line) {
|
||
|
|
// Need id + gtin; company is field index 4 in this dump schema:
|
||
|
|
// id, gtin, feed_id, feed_ids, company_id, raw_data, ...
|
||
|
|
fields := parseMySQLTupleFieldsN(line, 5)
|
||
|
|
if len(fields) >= 5 {
|
||
|
|
id := fields[0]
|
||
|
|
if _, ok := want[id]; ok && fields[4] == legacyCompany {
|
||
|
|
gtin := strings.TrimSpace(nullish(fields[1]))
|
||
|
|
if gtin != "" {
|
||
|
|
out[id] = gtin
|
||
|
|
remaining--
|
||
|
|
}
|
||
|
|
}
|
||
|
|
}
|
||
|
|
}
|
||
|
|
if err == io.EOF {
|
||
|
|
break
|
||
|
|
}
|
||
|
|
if err != nil {
|
||
|
|
return nil, err
|
||
|
|
}
|
||
|
|
}
|
||
|
|
return out, nil
|
||
|
|
}
|
||
|
|
|
||
|
|
func dumpSectionEnded(line, table string) bool {
|
||
|
|
if strings.HasPrefix(line, "CREATE TABLE") || strings.HasPrefix(line, "UNLOCK TABLES") || strings.HasPrefix(line, "LOCK TABLES") {
|
||
|
|
return true
|
||
|
|
}
|
||
|
|
if strings.HasPrefix(line, "INSERT INTO `") && !strings.HasPrefix(line, "INSERT INTO `"+table+"`") {
|
||
|
|
return true
|
||
|
|
}
|
||
|
|
if strings.HasPrefix(line, "DROP TABLE") {
|
||
|
|
return true
|
||
|
|
}
|
||
|
|
return false
|
||
|
|
}
|
||
|
|
|
||
|
|
func looksLikeTupleLine(line string) bool {
|
||
|
|
s := strings.TrimLeft(line, " \t")
|
||
|
|
return strings.HasPrefix(s, "(")
|
||
|
|
}
|
||
|
|
|
||
|
|
func parseMySQLTupleFields(line string) []string {
|
||
|
|
return parseMySQLTupleFieldsN(line, 0)
|
||
|
|
}
|
||
|
|
|
||
|
|
// parseMySQLTupleFieldsN parses up to maxFields (0 = all) from the first (...) tuple on the line.
|
||
|
|
func parseMySQLTupleFieldsN(line string, maxFields int) []string {
|
||
|
|
start := strings.Index(line, "(")
|
||
|
|
if start < 0 {
|
||
|
|
return nil
|
||
|
|
}
|
||
|
|
body := line[start+1:]
|
||
|
|
var out []string
|
||
|
|
for i := 0; i < len(body); {
|
||
|
|
if maxFields > 0 && len(out) >= maxFields {
|
||
|
|
break
|
||
|
|
}
|
||
|
|
for i < len(body) && (body[i] == ' ' || body[i] == '\t' || body[i] == ',') {
|
||
|
|
i++
|
||
|
|
}
|
||
|
|
if i >= len(body) || body[i] == ')' {
|
||
|
|
break
|
||
|
|
}
|
||
|
|
if body[i] == '\'' {
|
||
|
|
i++
|
||
|
|
var b strings.Builder
|
||
|
|
for i < len(body) {
|
||
|
|
ch := body[i]
|
||
|
|
if ch == '\\' && i+1 < len(body) {
|
||
|
|
b.WriteByte(body[i+1])
|
||
|
|
i += 2
|
||
|
|
continue
|
||
|
|
}
|
||
|
|
if ch == '\'' {
|
||
|
|
if i+1 < len(body) && body[i+1] == '\'' {
|
||
|
|
b.WriteByte('\'')
|
||
|
|
i += 2
|
||
|
|
continue
|
||
|
|
}
|
||
|
|
i++
|
||
|
|
break
|
||
|
|
}
|
||
|
|
b.WriteByte(ch)
|
||
|
|
i++
|
||
|
|
}
|
||
|
|
out = append(out, b.String())
|
||
|
|
continue
|
||
|
|
}
|
||
|
|
j := i
|
||
|
|
for j < len(body) && body[j] != ',' && body[j] != ')' {
|
||
|
|
j++
|
||
|
|
}
|
||
|
|
out = append(out, strings.TrimSpace(body[i:j]))
|
||
|
|
i = j
|
||
|
|
}
|
||
|
|
return out
|
||
|
|
}
|
||
|
|
|
||
|
|
func parseDumpJobID(raw string) (uuid.UUID, error) {
|
||
|
|
raw = strings.TrimSpace(raw)
|
||
|
|
if id, err := uuid.Parse(raw); err == nil {
|
||
|
|
return id, nil
|
||
|
|
}
|
||
|
|
// Legacy dump mixes UUID and numeric string PKs — keep remaps stable (migrator parity).
|
||
|
|
return uuid.NewSHA1(uuid.NameSpaceOID, []byte("processing_job:"+raw)), nil
|
||
|
|
}
|
||
|
|
|
||
|
|
func nullish(s string) string {
|
||
|
|
s = strings.TrimSpace(s)
|
||
|
|
if s == "" || strings.EqualFold(s, "NULL") {
|
||
|
|
return ""
|
||
|
|
}
|
||
|
|
return s
|
||
|
|
}
|
||
|
|
|
||
|
|
func atoiDefault(s string, def int) int {
|
||
|
|
s = nullish(s)
|
||
|
|
if s == "" {
|
||
|
|
return def
|
||
|
|
}
|
||
|
|
n, err := strconv.Atoi(s)
|
||
|
|
if err != nil {
|
||
|
|
return def
|
||
|
|
}
|
||
|
|
return n
|
||
|
|
}
|
||
|
|
|
||
|
|
func parseDumpTime(s string) time.Time {
|
||
|
|
s = nullish(s)
|
||
|
|
if s == "" {
|
||
|
|
return time.Time{}
|
||
|
|
}
|
||
|
|
for _, layout := range []string{
|
||
|
|
"2006-01-02 15:04:05",
|
||
|
|
time.RFC3339,
|
||
|
|
"2006-01-02 15:04:05.000",
|
||
|
|
} {
|
||
|
|
if t, err := time.ParseInLocation(layout, s, time.UTC); err == nil {
|
||
|
|
return t.UTC()
|
||
|
|
}
|
||
|
|
}
|
||
|
|
return time.Time{}
|
||
|
|
}
|
||
|
|
|
||
|
|
func parseDumpTimePtr(s string) *time.Time {
|
||
|
|
t := parseDumpTime(s)
|
||
|
|
if t.IsZero() {
|
||
|
|
return nil
|
||
|
|
}
|
||
|
|
return &t
|
||
|
|
}
|
||
|
|
|
||
|
|
func normalizeProcessingJobStatus(raw string) string {
|
||
|
|
switch strings.ToLower(strings.TrimSpace(raw)) {
|
||
|
|
case "completed", "success", "done":
|
||
|
|
return "completed"
|
||
|
|
case "failed", "error":
|
||
|
|
return "failed"
|
||
|
|
case "cancelled", "canceled", "skipped":
|
||
|
|
return "cancelled"
|
||
|
|
case "running", "processing":
|
||
|
|
return "running"
|
||
|
|
case "pending", "queued":
|
||
|
|
return "pending"
|
||
|
|
default:
|
||
|
|
return "failed"
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
func normalizeJobProductStatus(raw string) string {
|
||
|
|
switch strings.ToLower(strings.TrimSpace(raw)) {
|
||
|
|
case "processed", "completed", "success", "done":
|
||
|
|
return "processed"
|
||
|
|
case "failed", "error":
|
||
|
|
return "failed"
|
||
|
|
case "cancelled", "canceled", "skipped":
|
||
|
|
return "cancelled"
|
||
|
|
case "processing", "running":
|
||
|
|
return "processing"
|
||
|
|
case "pending", "queued":
|
||
|
|
return "pending"
|
||
|
|
default:
|
||
|
|
return "failed"
|
||
|
|
}
|
||
|
|
}
|