226 lines
10 KiB
PL/PgSQL
226 lines
10 KiB
PL/PgSQL
-- Backfill actionable notification lifecycle state and business-object links.
|
|
-- This migration is deliberately restricted to the verified tjwater_v2 BizDB.
|
|
|
|
BEGIN;
|
|
SET LOCAL lock_timeout = '5s';
|
|
|
|
DO $$
|
|
BEGIN
|
|
IF current_database() <> 'tjwater_v2' THEN
|
|
RAISE EXCEPTION 'Notification lifecycle migration is restricted to tjwater_v2';
|
|
END IF;
|
|
END $$;
|
|
|
|
LOCK TABLE workbench.operational_task_notifications IN SHARE ROW EXCLUSIVE MODE;
|
|
|
|
CREATE TEMP TABLE notification_lifecycle_projects ON COMMIT DROP AS
|
|
SELECT project_id, count(*)::bigint AS notification_count
|
|
FROM workbench.operational_task_notifications
|
|
GROUP BY project_id;
|
|
|
|
WITH mapped AS (
|
|
SELECT
|
|
notification.project_id,
|
|
notification.notification_id,
|
|
action.action_id,
|
|
action.approval_status,
|
|
action.execution_status
|
|
FROM workbench.operational_task_notifications AS notification
|
|
LEFT JOIN LATERAL (
|
|
SELECT candidate.action_id, candidate.approval_status, candidate.execution_status
|
|
FROM workbench.operational_task_actions AS candidate
|
|
WHERE candidate.project_id = notification.project_id
|
|
AND candidate.task_id = notification.task_id
|
|
AND 1 = (
|
|
SELECT count(*)
|
|
FROM workbench.operational_task_actions AS sibling
|
|
WHERE sibling.project_id = notification.project_id
|
|
AND sibling.task_id = notification.task_id
|
|
)
|
|
LIMIT 1
|
|
) AS action ON true
|
|
WHERE notification.document->>'kind' IN (
|
|
'action_pending', 'action_approval_pending', 'external_coordination_failed'
|
|
)
|
|
AND NOT (notification.document ? 'lifecycle_status')
|
|
)
|
|
UPDATE workbench.operational_task_notifications AS notification
|
|
SET document = notification.document || jsonb_strip_nulls(jsonb_build_object(
|
|
'lifecycle_status', CASE
|
|
WHEN mapped.action_id IS NULL THEN 'active'
|
|
WHEN notification.document->>'kind' IN ('action_pending', 'action_approval_pending')
|
|
AND mapped.approval_status = 'pending' THEN 'active'
|
|
WHEN notification.document->>'kind' = 'external_coordination_failed'
|
|
AND mapped.execution_status IN ('failed', 'unsupported', 'unknown') THEN 'active'
|
|
ELSE 'resolved'
|
|
END,
|
|
'subject_type', CASE WHEN mapped.action_id IS NULL THEN 'task' ELSE 'action' END,
|
|
'subject_id', COALESCE(mapped.action_id::text, notification.task_id::text),
|
|
'resolved_at', CASE
|
|
WHEN (notification.document->>'kind' IN ('action_pending', 'action_approval_pending')
|
|
AND mapped.approval_status <> 'pending')
|
|
OR (notification.document->>'kind' = 'external_coordination_failed'
|
|
AND mapped.execution_status NOT IN ('failed', 'unsupported', 'unknown'))
|
|
THEN COALESCE(notification.document->'resolved_at', to_jsonb(CURRENT_TIMESTAMP))
|
|
ELSE NULL
|
|
END,
|
|
'resolution_reason', CASE
|
|
WHEN notification.document->>'resolution_reason' IS NOT NULL
|
|
THEN notification.document->>'resolution_reason'
|
|
WHEN notification.document->>'kind' IN ('action_pending', 'action_approval_pending')
|
|
AND mapped.approval_status <> 'pending' THEN 'migration_action_decided'
|
|
WHEN notification.document->>'kind' = 'external_coordination_failed'
|
|
AND mapped.execution_status NOT IN ('failed', 'unsupported', 'unknown')
|
|
THEN 'migration_external_coordination_recovered'
|
|
ELSE NULL
|
|
END
|
|
))
|
|
FROM mapped
|
|
WHERE notification.project_id = mapped.project_id
|
|
AND notification.notification_id = mapped.notification_id;
|
|
|
|
DO $block$
|
|
BEGIN
|
|
IF to_regclass('workbench.operational_task_access_requests') IS NOT NULL THEN
|
|
EXECUTE $update$
|
|
WITH mapped AS (
|
|
SELECT
|
|
notification.project_id,
|
|
notification.notification_id,
|
|
access_request.access_request_id,
|
|
access_request.status
|
|
FROM workbench.operational_task_notifications AS notification
|
|
LEFT JOIN LATERAL (
|
|
SELECT candidate.access_request_id, candidate.status
|
|
FROM workbench.operational_task_access_requests AS candidate
|
|
WHERE candidate.project_id = notification.project_id
|
|
AND candidate.task_id = notification.task_id
|
|
AND 1 = (
|
|
SELECT count(*)
|
|
FROM workbench.operational_task_access_requests AS sibling
|
|
WHERE sibling.project_id = notification.project_id
|
|
AND sibling.task_id = notification.task_id
|
|
)
|
|
LIMIT 1
|
|
) AS access_request ON true
|
|
WHERE notification.document->>'kind' = 'access_approval_pending'
|
|
AND NOT (notification.document ? 'lifecycle_status')
|
|
)
|
|
UPDATE workbench.operational_task_notifications AS notification
|
|
SET document = notification.document || jsonb_strip_nulls(jsonb_build_object(
|
|
'lifecycle_status', CASE
|
|
WHEN mapped.access_request_id IS NULL OR mapped.status = 'pending'
|
|
THEN 'active' ELSE 'resolved'
|
|
END,
|
|
'subject_type', CASE
|
|
WHEN mapped.access_request_id IS NULL THEN 'task'
|
|
ELSE 'access_request'
|
|
END,
|
|
'subject_id', COALESCE(
|
|
mapped.access_request_id::text,
|
|
notification.task_id::text
|
|
),
|
|
'resolved_at', CASE
|
|
WHEN mapped.access_request_id IS NOT NULL AND mapped.status <> 'pending'
|
|
THEN COALESCE(
|
|
notification.document->'resolved_at',
|
|
to_jsonb(CURRENT_TIMESTAMP)
|
|
)
|
|
ELSE NULL
|
|
END,
|
|
'resolution_reason', CASE
|
|
WHEN notification.document->>'resolution_reason' IS NOT NULL
|
|
THEN notification.document->>'resolution_reason'
|
|
WHEN mapped.status <> 'pending' THEN 'migration_access_request_decided'
|
|
ELSE NULL
|
|
END
|
|
))
|
|
FROM mapped
|
|
WHERE notification.project_id = mapped.project_id
|
|
AND notification.notification_id = mapped.notification_id
|
|
$update$;
|
|
END IF;
|
|
END $block$;
|
|
|
|
UPDATE workbench.operational_task_notifications AS notification
|
|
SET document = notification.document || jsonb_strip_nulls(jsonb_build_object(
|
|
'lifecycle_status', CASE
|
|
WHEN notification.document->>'kind' = 'task_failed'
|
|
AND COALESCE(task.document->>'lifecycle_status', task.document->>'status') = 'failed'
|
|
THEN 'active'
|
|
WHEN notification.document->>'kind' = 'human_input_required'
|
|
AND COALESCE(task.document->>'lifecycle_status', task.document->>'status') = 'waiting_human'
|
|
THEN 'active'
|
|
WHEN notification.document->>'kind' IN ('task_failed', 'human_input_required')
|
|
THEN 'resolved'
|
|
ELSE COALESCE(notification.document->>'lifecycle_status', 'active')
|
|
END,
|
|
'subject_type', COALESCE(notification.document->>'subject_type', 'task'),
|
|
'subject_id', COALESCE(notification.document->>'subject_id', notification.task_id::text),
|
|
'resolved_at', CASE
|
|
WHEN notification.document->>'kind' = 'task_failed'
|
|
AND COALESCE(task.document->>'lifecycle_status', task.document->>'status') <> 'failed'
|
|
THEN COALESCE(notification.document->'resolved_at', to_jsonb(CURRENT_TIMESTAMP))
|
|
WHEN notification.document->>'kind' = 'human_input_required'
|
|
AND COALESCE(task.document->>'lifecycle_status', task.document->>'status') <> 'waiting_human'
|
|
THEN COALESCE(notification.document->'resolved_at', to_jsonb(CURRENT_TIMESTAMP))
|
|
ELSE notification.document->'resolved_at'
|
|
END,
|
|
'resolution_reason', CASE
|
|
WHEN notification.document->>'kind' = 'task_failed'
|
|
AND COALESCE(task.document->>'lifecycle_status', task.document->>'status') <> 'failed'
|
|
THEN 'migration_task_recovered'
|
|
WHEN notification.document->>'kind' = 'human_input_required'
|
|
AND COALESCE(task.document->>'lifecycle_status', task.document->>'status') <> 'waiting_human'
|
|
THEN 'migration_human_input_completed'
|
|
ELSE notification.document->>'resolution_reason'
|
|
END
|
|
))
|
|
FROM workbench.operational_tasks AS task
|
|
WHERE notification.project_id = task.project_id
|
|
AND notification.task_id = task.task_id
|
|
AND notification.document->>'kind' NOT IN (
|
|
'action_pending', 'action_approval_pending',
|
|
'external_coordination_failed', 'access_approval_pending'
|
|
)
|
|
AND NOT (notification.document ? 'lifecycle_status');
|
|
|
|
WITH inserted_events AS (
|
|
INSERT INTO workbench.domain_event_log (
|
|
project_id,
|
|
event_id,
|
|
topic,
|
|
aggregate_type,
|
|
aggregate_id,
|
|
aggregate_revision,
|
|
metadata
|
|
)
|
|
SELECT
|
|
project_id,
|
|
(
|
|
substr(md5(project_id::text || ':notification-lifecycle-v1'), 1, 8)
|
|
|| '-' || substr(md5(project_id::text || ':notification-lifecycle-v1'), 9, 4)
|
|
|| '-5' || substr(md5(project_id::text || ':notification-lifecycle-v1'), 14, 3)
|
|
|| '-a' || substr(md5(project_id::text || ':notification-lifecycle-v1'), 18, 3)
|
|
|| '-' || substr(md5(project_id::text || ':notification-lifecycle-v1'), 21, 12)
|
|
)::uuid,
|
|
'operational.notification.changed',
|
|
'operational_notification_batch',
|
|
project_id::text,
|
|
1,
|
|
jsonb_build_object(
|
|
'change', 'lifecycle_backfilled',
|
|
'notification_count', notification_count
|
|
)
|
|
FROM notification_lifecycle_projects
|
|
ON CONFLICT (project_id, event_id) DO NOTHING
|
|
RETURNING project_id, event_seq
|
|
)
|
|
SELECT pg_notify('workbench_events', project_id::text || ':' || event_seq::text)
|
|
FROM inserted_events;
|
|
|
|
COMMENT ON COLUMN workbench.operational_task_notifications.document IS
|
|
'通知内容及 active/resolved 生命周期；用户 unread/read 状态独立存于 notification_reads。';
|
|
|
|
COMMIT;
|