Merge remote-tracking branch 'origin/master'

svc/product_series_auto_scheduler.go

@@ -248,6 +248,11 @@ WHERE NOT (row_key = ANY($1))
 		}
 		stats.Changed++
 	}
+	retryCount, err := productSeriesEnqueueUnmatchedStateTx(ctx, tx, reason)
+	if err != nil {
+		return stats, err
+	}
+	stats.Changed += retryCount
 
 	return stats, tx.Commit()
 }
@@ -687,6 +692,63 @@ ON CONFLICT (row_key) WHERE status IN ('pending','processing') DO NOTHING
 	return 1, nil
 }
 
+func productSeriesEnqueueUnmatchedStateTx(ctx context.Context, tx *sql.Tx, reason string) (int, error) {
+	rows, err := tx.QueryContext(ctx, `
+WITH candidates AS (
+  SELECT
+    s.row_key,
+    s.product_code,
+    s.color_code,
+    s.dim3_code
+  FROM mk_product_series_stock_state s
+  LEFT JOIN mmitem m
+    ON BTRIM(m.code) = BTRIM(s.product_code)
+  LEFT JOIN mk_dim_token_map d1
+    ON d1.dim_column='dimval1'
+   AND BTRIM(d1.token) = BTRIM(s.color_code)
+  LEFT JOIN mk_dim_token_map d3
+    ON d3.dim_column='dimval3'
+   AND BTRIM(d3.token) = BTRIM(s.dim3_code)
+  LEFT JOIN zbggseri z
+    ON z.mmitem_id = m.id
+   AND z.dim1 = d1.dim_id
+   AND (
+     (NULLIF(BTRIM(s.dim3_code), '') IS NULL AND z.dim3 IS NULL)
+     OR z.dim3 = d3.dim_id
+   )
+  WHERE s.total_qty > 0
+    AND s.last_seen_at > now() - interval '30 days'
+    AND (
+      m.id IS NULL
+      OR d1.dim_id IS NULL
+      OR (NULLIF(BTRIM(s.dim3_code), '') IS NOT NULL AND d3.dim_id IS NULL)
+      OR z.id IS NULL
+    )
+  ORDER BY s.updated_at DESC
+  LIMIT 2000
+),
+inserted AS (
+  INSERT INTO mk_product_series_recalc_queue (row_key, product_code, color_code, dim3_code, reason, status, attempts, available_at, queued_at, last_error, created_at, updated_at)
+  SELECT row_key, product_code, color_code, dim3_code, $1, 'pending', 0, now(), now(), '', now(), now()
+  FROM candidates
+  ON CONFLICT (row_key) WHERE status IN ('pending','processing') DO NOTHING
+  RETURNING id
+)
+SELECT COUNT(*) FROM inserted
+`, "unmatched_retry_"+strings.TrimSpace(reason))
+	if err != nil {
+		return 0, err
+	}
+	defer rows.Close()
+	var count int
+	if rows.Next() {
+		if err := rows.Scan(&count); err != nil {
+			return 0, err
+		}
+	}
+	return count, rows.Err()
+}
+
 func productSeriesMarkQueueDone(ctx context.Context, pg *sql.DB, id int64) error {
 	_, err := pg.ExecContext(ctx, `
 UPDATE mk_product_series_recalc_queue