No run has completed yet.
;
- }
-
return (
+
Initiate merges
-
-
- {loading ? "Running…" : "Start run"}
-
+
+
+
+ Re-run (ignore a recent run)
+
+
+
- {error ? (
-
- ) : null}
-
-
- Run result
-
-
+ {job.error && (
+
+ {job.error}
-
+ )}
+
+ {revealed && (
+
+ Runs
+
+ record.state === "succeeded" ? (
+
+ {record.error || "Run failed."}
+
+ ) : (
+ Running…
+ )
+ }
+ />
+
+ )}
- {state}
-
- );
-}
-
-async function parseJsonResponse(response) {
- const text = await response.text();
-
- try {
- return text ? JSON.parse(text) : null;
- } catch {
- return { raw: text };
- }
-}
-
-function EtlPanel({ title, description, baseUrl, fields, getToken }) {
+function EtlPanel({ title, description, baseUrl, fields, fetcher }) {
const [values, setValues] = useState(() =>
Object.fromEntries(
fields.map((field) => [
@@ -82,17 +60,20 @@ function EtlPanel({ title, description, baseUrl, fields, getToken }) {
]),
),
);
- const [starting, setStarting] = useState(false);
- const [refreshing, setRefreshing] = useState(false);
- const [record, setRecord] = useState(null);
- const [error, setError] = useState(null);
+ const [forceRerun, setForceRerun] = useState(false);
+ const [revealed, setRevealed] = useState(false);
+
+ // Look up this pipeline's runs with default params (a stable key, so editing
+ // the tunable fields doesn't refetch on every keystroke). The pipeline is its
+ // own service, so /find returns its runs per the backend's dedup policy.
+ const job = useJobRunner({ baseUrl, fetcher, lookupBody: {} });
function setField(key, value) {
setValues((prev) => ({ ...prev, [key]: value }));
}
function buildRequestBody() {
- const body = {};
+ const body = { force: forceRerun };
for (const field of fields) {
const value = values[field.key];
@@ -107,81 +88,22 @@ function EtlPanel({ title, description, baseUrl, fields, getToken }) {
return body;
}
- async function handleStart() {
- setStarting(true);
- setError(null);
-
- try {
- const token = await getToken();
-
- const response = await fetch(`${baseUrl}/start`, {
- method: "POST",
- headers: {
- "Content-Type": "application/json",
- Authorization: `Bearer ${token}`,
- },
- body: JSON.stringify(buildRequestBody()),
- });
-
- const parsed = await parseJsonResponse(response);
-
- if (response.status === 409) {
- setError({
- status: 409,
- message: "A run is already in progress — refresh status to check.",
- body: parsed,
- });
- return;
- }
-
- if (!response.ok) {
- setError({ status: response.status, body: parsed });
- return;
- }
-
- setRecord(parsed);
- } catch (err) {
- setError({
- status: null,
- body: err instanceof Error ? err.message : String(err),
- });
- } finally {
- setStarting(false);
+ async function handleTrigger() {
+ // A recent run exists and we're not forcing → just reveal it.
+ if (job.hasExisting && !forceRerun && !revealed) {
+ setRevealed(true);
+ return;
}
+ setRevealed(true);
+ await job.start(buildRequestBody());
+ setForceRerun(false);
}
- async function handleRefresh() {
- setRefreshing(true);
- setError(null);
-
- try {
- // GET /status is unauthenticated per the ETL OpenAPI spec.
- const response = await fetch(`${baseUrl}/status`);
- const parsed = await parseJsonResponse(response);
-
- if (!response.ok) {
- setError({ status: response.status, body: parsed });
- return;
- }
-
- setRecord(parsed);
- } catch (err) {
- setError({
- status: null,
- body: err instanceof Error ? err.message : String(err),
- });
- } finally {
- setRefreshing(false);
- }
- }
-
- const isRunning = record?.state === "running";
-
return (
{title}
- {description}
@@ -194,6 +116,7 @@ function EtlPanel({ title, description, baseUrl, fields, getToken }) {
type="checkbox"
checked={values[field.key]}
onChange={(e) => setField(field.key, e.target.checked)}
+ disabled={job.isRunning}
className="accent-primary"
/>
{field.label}
@@ -209,6 +132,7 @@ function EtlPanel({ title, description, baseUrl, fields, getToken }) {
onChange={(e) => setField(field.key, e.target.value)}
placeholder={field.placeholder}
min={field.type === "number" ? 1 : undefined}
+ disabled={job.isRunning}
className="w-full"
/>
@@ -220,43 +144,62 @@ function EtlPanel({ title, description, baseUrl, fields, getToken }) {
)}
-
-
+
- {starting ? "Starting…" : "Start run"}
-
-
- {refreshing ? "Refreshing…" : "Refresh status"}
-
+ />
+
+ Re-run (ignore a recent run)
+
- {error ? (
+ {job.error ? (
- {error.message ? (
-
{error.message}
- ) : null}
-
+ {job.error}
) : null}
-
Run status
- {record ? (
-
+
Runs
+ {revealed && job.records.length ? (
+
+ record.state === "succeeded" ? (
+
+ {record.error || "Run failed."}
+
+ ) : (
+ Running…
+ )
+ }
+ />
) : (
- No run started yet. Start a run or refresh to fetch the latest
- status.
+ No run started yet. Start a run to begin.
)}
@@ -267,11 +210,7 @@ function EtlPanel({ title, description, baseUrl, fields, getToken }) {
export default function EtlPage() {
const config = useConfig();
const { getAccessTokenSilently } = useAuth0();
-
- const getToken = () =>
- getAccessTokenSilently({
- authorizationParams: { audience: config.auth0.audience },
- });
+ const fetcher = createAuthenticatedFetcher(config, getAccessTokenSilently);
return (
@@ -281,8 +220,8 @@ export default function EtlPage() {
ETL Pipelines
- Start an ETL run and check its status. Only one run per pipeline may
- be active at a time.
+ Start an ETL run and watch its status. A recent run for a pipeline is
+ shown rather than started again; use “Re-run” to force a fresh run.
@@ -292,14 +231,14 @@ export default function EtlPage() {
description="Load GEM oil & gas data from the configured spreadsheet and post it to Stitch."
baseUrl={config.etlGemBaseUrl}
fields={GEM_FIELDS}
- getToken={getToken}
+ fetcher={fetcher}
/>