From ccf24a6525201df2e844106ba2f9384630c8e4d8 Mon Sep 17 00:00:00 2001 From: Nikhil <118773738+pablohashescobar@users.noreply.github.com> Date: Tue, 12 Aug 2025 19:49:51 +0530 Subject: [PATCH] [WEB-4667] chore: outbox event payload and change the way transaction variable is set (#3888) --- apps/api/plane/bgtasks/data_import_task.py | 4 +- .../event_stream/bgtasks/outbox_cleaner.py | 2 +- ...y_issue_assignee_outbox_insert_and_more.py | 87 +++++ apps/api/plane/event_stream/models/issue.py | 314 +++++++++--------- apps/api/plane/settings/common.py | 2 + .../bulk_update_issue_relations_task.py | 8 +- 6 files changed, 247 insertions(+), 170 deletions(-) create mode 100644 apps/api/plane/event_stream/migrations/0003_remove_issueassigneeproxy_issue_assignee_outbox_insert_and_more.py diff --git a/apps/api/plane/bgtasks/data_import_task.py b/apps/api/plane/bgtasks/data_import_task.py index e7c90b9f54..a8789f0800 100644 --- a/apps/api/plane/bgtasks/data_import_task.py +++ b/apps/api/plane/bgtasks/data_import_task.py @@ -157,7 +157,9 @@ def process_single_issue(slug, project, user_id, issue_data): try: with transaction.atomic(): with connection.cursor() as cur: - cur.execute("SET LOCAL plane.initiator_type = 'SYSTEM.IMPORT'") + cur.execute( + "SELECT set_config('plane.initiator_type', 'SYSTEM.IMPORT', true)" + ) # Process the main issue issue_data = sanitize_issue_data(issue_data) serializer = IssueSerializer( diff --git a/apps/api/plane/event_stream/bgtasks/outbox_cleaner.py b/apps/api/plane/event_stream/bgtasks/outbox_cleaner.py index 6f27a3b80d..e578fe3247 100644 --- a/apps/api/plane/event_stream/bgtasks/outbox_cleaner.py +++ b/apps/api/plane/event_stream/bgtasks/outbox_cleaner.py @@ -95,7 +95,7 @@ def delete_outbox_records() -> None: mongo_available = False # Calculate cutoff time - cutoff_days = int(os.environ.get("OUTBOX_CLEANER_CUTOFF_DAYS", 2)) + cutoff_days = int(os.environ.get("OUTBOX_CLEANER_CUTOFF_DAYS", 7)) cutoff_time = timezone.now() - timedelta(days=cutoff_days) logger.info( f"Processing outbox records older than {cutoff_time} (cutoff: {cutoff_days} days)" diff --git a/apps/api/plane/event_stream/migrations/0003_remove_issueassigneeproxy_issue_assignee_outbox_insert_and_more.py b/apps/api/plane/event_stream/migrations/0003_remove_issueassigneeproxy_issue_assignee_outbox_insert_and_more.py new file mode 100644 index 0000000000..42647fb5aa --- /dev/null +++ b/apps/api/plane/event_stream/migrations/0003_remove_issueassigneeproxy_issue_assignee_outbox_insert_and_more.py @@ -0,0 +1,87 @@ +# Generated by Django 4.2.22 on 2025-08-11 13:57 + +from django.db import migrations +import pgtrigger.compiler +import pgtrigger.migrations + + +class Migration(migrations.Migration): + + dependencies = [ + ("event_stream", "0002_alter_outbox_initiator_id"), + ] + + operations = [ + pgtrigger.migrations.RemoveTrigger( + model_name="issueassigneeproxy", + name="issue_assignee_outbox_insert", + ), + pgtrigger.migrations.RemoveTrigger( + model_name="issueassigneeproxy", + name="issue_assignee_outbox_update", + ), + pgtrigger.migrations.RemoveTrigger( + model_name="issuelabelproxy", + name="issue_label_outbox_insert", + ), + pgtrigger.migrations.RemoveTrigger( + model_name="issuelabelproxy", + name="issue_label_outbox_update", + ), + pgtrigger.migrations.AddTrigger( + model_name="issueassigneeproxy", + trigger=pgtrigger.compiler.Trigger( + name="issue_assignee_outbox_insert", + sql=pgtrigger.compiler.UpsertTriggerSql( + func="\n DECLARE\n event_type_name TEXT;\n enriched_data JSONB;\n assignee_ids JSONB;\n previous_assignee_ids JSONB;\n label_ids JSONB;\n issue_data JSONB;\n BEGIN\n -- Determine event type based on issue type\n SELECT \n CASE \n WHEN i.type_id IS NULL THEN 'issue.assignee.added'\n WHEN it.is_epic = true THEN 'epic.assignee.added'\n ELSE 'issue.assignee.added'\n END\n INTO event_type_name\n FROM issues i\n LEFT JOIN issue_types it ON it.id = i.type_id\n WHERE i.id = NEW.issue_id;\n \n -- If no issue found, default to 'issue.assignee.added'\n IF event_type_name IS NULL THEN\n event_type_name := 'issue.assignee.added';\n END IF;\n \n -- Get the complete issue data (excluding description fields)\n SELECT to_jsonb(i) - 'description_html' - 'description_binary' - 'description' - 'description_stripped'\n INTO issue_data\n FROM issues i\n WHERE i.id = NEW.issue_id;\n \n -- Fetch ALL current assignee IDs (including the newly added one)\n SELECT COALESCE(jsonb_agg(ia.assignee_id ORDER BY ia.created_at), '[]'::jsonb)\n INTO assignee_ids\n FROM issue_assignees ia\n WHERE ia.issue_id = NEW.issue_id AND ia.deleted_at IS NULL;\n \n -- Fetch previous assignee IDs (excluding the newly added one)\n SELECT COALESCE(jsonb_agg(ia.assignee_id ORDER BY ia.created_at), '[]'::jsonb)\n INTO previous_assignee_ids\n FROM issue_assignees ia\n WHERE ia.issue_id = NEW.issue_id AND ia.deleted_at IS NULL AND ia.id != NEW.id;\n \n -- Fetch label IDs\n SELECT COALESCE(jsonb_agg(il.label_id ORDER BY il.created_at), '[]'::jsonb)\n INTO label_ids\n FROM issue_labels il\n WHERE il.issue_id = NEW.issue_id AND il.deleted_at IS NULL;\n \n -- Create enriched data with complete issue, ALL assignee IDs (including new one), and label IDs\n enriched_data := issue_data || jsonb_build_object(\n 'assignee_ids', assignee_ids,\n 'label_ids', label_ids\n );\n \n -- Create previous attributes with only the previous assignee IDs\n BEGIN\n INSERT INTO outbox (event_id, event_type, entity_type, entity_id, workspace_id, project_id, payload, created_at, initiator_id, initiator_type)\n VALUES (\n gen_random_uuid(),\n event_type_name,\n 'issue_assignee',\n NEW.issue_id,\n NEW.workspace_id,\n NEW.project_id,\n jsonb_build_object('data', enriched_data, 'previous_attributes', jsonb_build_object('assignee_ids', previous_assignee_ids)),\n now(),\n NEW.created_by_id,\n COALESCE(current_setting('plane.initiator_type', true), 'USER')\n )\n ON CONFLICT DO NOTHING;\n EXCEPTION\n WHEN others THEN\n RAISE WARNING 'Outbox insert failed for issue_assignee %, reason: %',\n NEW.issue_id, SQLERRM;\n END;\n RETURN NEW;\n END;\n ", + hash="fb9078084f95a0bcc244f461ab1a4660e161b4de", + operation="INSERT", + pgid="pgtrigger_issue_assignee_outbox_insert_51e4c", + table="issue_assignees", + when="AFTER", + ), + ), + ), + pgtrigger.migrations.AddTrigger( + model_name="issueassigneeproxy", + trigger=pgtrigger.compiler.Trigger( + name="issue_assignee_outbox_update", + sql=pgtrigger.compiler.UpsertTriggerSql( + func="\n DECLARE\n event_type_name TEXT;\n enriched_data JSONB;\n current_assignee_ids JSONB;\n previous_assignee_ids JSONB;\n label_ids JSONB;\n issue_data JSONB;\n BEGIN\n -- Only trigger on soft delete (deleted_at changing from NULL to NOT NULL)\n IF OLD.deleted_at IS NULL AND NEW.deleted_at IS NOT NULL THEN\n -- Determine event type based on issue type\n SELECT \n CASE \n WHEN i.type_id IS NULL THEN 'issue.assignee.removed'\n WHEN it.is_epic = true THEN 'epic.assignee.removed'\n ELSE 'issue.assignee.removed'\n END\n INTO event_type_name\n FROM issues i\n LEFT JOIN issue_types it ON it.id = i.type_id\n WHERE i.id = OLD.issue_id;\n \n -- If no issue found, default to 'issue.assignee.removed'\n IF event_type_name IS NULL THEN\n event_type_name := 'issue.assignee.removed';\n END IF;\n \n -- Get the complete issue data (excluding description fields)\n SELECT to_jsonb(i) - 'description_html' - 'description_binary' - 'description' - 'description_stripped'\n INTO issue_data\n FROM issues i\n WHERE i.id = OLD.issue_id;\n \n -- Fetch current assignee IDs (excluding the one being removed)\n SELECT COALESCE(jsonb_agg(ia.assignee_id ORDER BY ia.created_at), '[]'::jsonb)\n INTO current_assignee_ids\n FROM issue_assignees ia\n WHERE ia.issue_id = OLD.issue_id AND ia.deleted_at IS NULL AND ia.id != OLD.id;\n \n -- Fetch ALL assignee IDs (including the one being removed) for previous_attributes\n SELECT COALESCE(jsonb_agg(ia.assignee_id ORDER BY ia.created_at), '[]'::jsonb)\n INTO previous_assignee_ids\n FROM issue_assignees ia\n WHERE ia.issue_id = OLD.issue_id AND ia.deleted_at IS NULL;\n \n -- Fetch label IDs\n SELECT COALESCE(jsonb_agg(il.label_id ORDER BY il.created_at), '[]'::jsonb)\n INTO label_ids\n FROM issue_labels il\n WHERE il.issue_id = OLD.issue_id AND il.deleted_at IS NULL;\n \n -- Create enriched data with complete issue and current assignee IDs (after removal)\n enriched_data := issue_data || jsonb_build_object(\n 'assignee_ids', current_assignee_ids,\n 'label_ids', label_ids\n );\n \n -- Create previous attributes with only the previous assignee IDs\n BEGIN\n INSERT INTO outbox (event_id, event_type, entity_type, entity_id, workspace_id, project_id, payload, created_at, initiator_id, initiator_type)\n VALUES (\n gen_random_uuid(),\n event_type_name,\n 'issue_assignee',\n OLD.issue_id,\n OLD.workspace_id,\n OLD.project_id,\n jsonb_build_object('data', enriched_data, 'previous_attributes', jsonb_build_object('assignee_ids', previous_assignee_ids)),\n now(),\n NEW.updated_by_id,\n COALESCE(current_setting('plane.initiator_type', true), 'USER')\n )\n ON CONFLICT DO NOTHING;\n EXCEPTION\n WHEN others THEN\n RAISE WARNING 'Outbox delete-event failed for issue_assignee %, reason: %',\n OLD.issue_id, SQLERRM;\n END;\n END IF;\n RETURN NEW;\n END;\n ", + hash="5431a3c2730df15b182fa5344f836d49789827fa", + operation="UPDATE", + pgid="pgtrigger_issue_assignee_outbox_update_97b55", + table="issue_assignees", + when="BEFORE", + ), + ), + ), + pgtrigger.migrations.AddTrigger( + model_name="issuelabelproxy", + trigger=pgtrigger.compiler.Trigger( + name="issue_label_outbox_insert", + sql=pgtrigger.compiler.UpsertTriggerSql( + func="\n DECLARE\n event_type_name TEXT;\n enriched_data JSONB;\n assignee_ids JSONB;\n label_ids JSONB;\n previous_label_ids JSONB;\n issue_data JSONB;\n BEGIN\n -- Determine event type based on issue type\n SELECT \n CASE \n WHEN i.type_id IS NULL THEN 'issue.label.added'\n WHEN it.is_epic = true THEN 'epic.label.added'\n ELSE 'issue.label.added'\n END\n INTO event_type_name\n FROM issues i\n LEFT JOIN issue_types it ON it.id = i.type_id\n WHERE i.id = NEW.issue_id;\n \n -- If no issue found, default to 'issue.label.added'\n IF event_type_name IS NULL THEN\n event_type_name := 'issue.label.added';\n END IF;\n \n -- Get the complete issue data (excluding description fields)\n SELECT to_jsonb(i) - 'description_html' - 'description_binary' - 'description' - 'description_stripped'\n INTO issue_data\n FROM issues i\n WHERE i.id = NEW.issue_id;\n \n -- Fetch assignee IDs\n SELECT COALESCE(jsonb_agg(ia.assignee_id ORDER BY ia.created_at), '[]'::jsonb)\n INTO assignee_ids\n FROM issue_assignees ia\n WHERE ia.issue_id = NEW.issue_id AND ia.deleted_at IS NULL;\n \n -- Fetch ALL current label IDs (including the newly added one)\n SELECT COALESCE(jsonb_agg(il.label_id ORDER BY il.created_at), '[]'::jsonb)\n INTO label_ids\n FROM issue_labels il\n WHERE il.issue_id = NEW.issue_id AND il.deleted_at IS NULL;\n \n -- Fetch previous label IDs (excluding the newly added one)\n SELECT COALESCE(jsonb_agg(il.label_id ORDER BY il.created_at), '[]'::jsonb)\n INTO previous_label_ids\n FROM issue_labels il\n WHERE il.issue_id = NEW.issue_id AND il.deleted_at IS NULL AND il.id != NEW.id;\n \n -- Create enriched data with complete issue, assignee IDs, and ALL label IDs (including new one)\n enriched_data := issue_data || jsonb_build_object(\n 'assignee_ids', assignee_ids,\n 'label_ids', label_ids\n );\n \n -- Create previous attributes with only the previous label IDs\n BEGIN\n INSERT INTO outbox (event_id, event_type, entity_type, entity_id, workspace_id, project_id, payload, created_at, initiator_id, initiator_type)\n VALUES (\n gen_random_uuid(),\n event_type_name,\n 'issue_label',\n NEW.issue_id,\n NEW.workspace_id,\n NEW.project_id,\n jsonb_build_object('data', enriched_data, 'previous_attributes', jsonb_build_object('label_ids', previous_label_ids)),\n now(),\n NEW.created_by_id,\n COALESCE(current_setting('plane.initiator_type', true), 'USER')\n )\n ON CONFLICT DO NOTHING;\n EXCEPTION\n WHEN others THEN\n RAISE WARNING 'Outbox insert failed for issue_label %, reason: %',\n NEW.issue_id, SQLERRM;\n END;\n RETURN NEW;\n END;\n ", + hash="8e775687be295714109f924ab6f7f54540ea0256", + operation="INSERT", + pgid="pgtrigger_issue_label_outbox_insert_a8b0d", + table="issue_labels", + when="AFTER", + ), + ), + ), + pgtrigger.migrations.AddTrigger( + model_name="issuelabelproxy", + trigger=pgtrigger.compiler.Trigger( + name="issue_label_outbox_update", + sql=pgtrigger.compiler.UpsertTriggerSql( + func="\n DECLARE\n event_type_name TEXT;\n enriched_data JSONB;\n assignee_ids JSONB;\n current_label_ids JSONB;\n previous_label_ids JSONB;\n issue_data JSONB;\n BEGIN\n -- Only trigger on soft delete (deleted_at changing from NULL to NOT NULL)\n IF OLD.deleted_at IS NULL AND NEW.deleted_at IS NOT NULL THEN\n -- Determine event type based on issue type\n SELECT \n CASE \n WHEN i.type_id IS NULL THEN 'issue.label.removed'\n WHEN it.is_epic = true THEN 'epic.label.removed'\n ELSE 'issue.label.removed'\n END\n INTO event_type_name\n FROM issues i\n LEFT JOIN issue_types it ON it.id = i.type_id\n WHERE i.id = OLD.issue_id;\n \n -- If no issue found, default to 'issue.label.removed'\n IF event_type_name IS NULL THEN\n event_type_name := 'issue.label.removed';\n END IF;\n \n -- Get the complete issue data (excluding description fields)\n SELECT to_jsonb(i) - 'description_html' - 'description_binary' - 'description' - 'description_stripped'\n INTO issue_data\n FROM issues i\n WHERE i.id = OLD.issue_id;\n \n -- Fetch assignee IDs\n SELECT COALESCE(jsonb_agg(ia.assignee_id ORDER BY ia.created_at), '[]'::jsonb)\n INTO assignee_ids\n FROM issue_assignees ia\n WHERE ia.issue_id = OLD.issue_id AND ia.deleted_at IS NULL;\n \n -- Fetch current label IDs (excluding the one being removed)\n SELECT COALESCE(jsonb_agg(il.label_id ORDER BY il.created_at), '[]'::jsonb)\n INTO current_label_ids\n FROM issue_labels il\n WHERE il.issue_id = OLD.issue_id AND il.deleted_at IS NULL AND il.id != OLD.id;\n \n -- Fetch ALL label IDs (including the one being removed) for previous_attributes\n SELECT COALESCE(jsonb_agg(il.label_id ORDER BY il.created_at), '[]'::jsonb)\n INTO previous_label_ids\n FROM issue_labels il\n WHERE il.issue_id = OLD.issue_id AND il.deleted_at IS NULL;\n \n -- Create enriched data with complete issue, assignee IDs, and current label IDs (after removal)\n enriched_data := issue_data || jsonb_build_object(\n 'assignee_ids', assignee_ids,\n 'label_ids', current_label_ids\n );\n \n -- Create previous attributes with only the previous label IDs (including the removed one)\n BEGIN\n INSERT INTO outbox (event_id, event_type, entity_type, entity_id, workspace_id, project_id, payload, created_at, initiator_id, initiator_type)\n VALUES (\n gen_random_uuid(),\n event_type_name,\n 'issue_label',\n OLD.issue_id,\n OLD.workspace_id,\n OLD.project_id,\n jsonb_build_object('data', enriched_data, 'previous_attributes', jsonb_build_object('label_ids', previous_label_ids)),\n now(),\n NEW.updated_by_id,\n COALESCE(current_setting('plane.initiator_type', true), 'USER')\n )\n ON CONFLICT DO NOTHING;\n EXCEPTION\n WHEN others THEN\n RAISE WARNING 'Outbox delete-event failed for issue_label %, reason: %',\n OLD.issue_id, SQLERRM;\n END;\n END IF;\n RETURN NEW;\n END;\n ", + hash="d95ed7ede7029dd8d4bb8e24f69377de71d6291d", + operation="UPDATE", + pgid="pgtrigger_issue_label_outbox_update_47e56", + table="issue_labels", + when="BEFORE", + ), + ), + ), + ] diff --git a/apps/api/plane/event_stream/models/issue.py b/apps/api/plane/event_stream/models/issue.py index 5163d046a8..b45bf85e17 100644 --- a/apps/api/plane/event_stream/models/issue.py +++ b/apps/api/plane/event_stream/models/issue.py @@ -399,12 +399,7 @@ class IssueAssigneeProxy(IssueAssignee): 'label_ids', label_ids ); - -- Create previous attributes with assignee IDs that existed before the addition - issue_data := issue_data || jsonb_build_object( - 'assignee_ids', previous_assignee_ids, - 'label_ids', label_ids - ); - + -- Create previous attributes with only the previous assignee IDs BEGIN INSERT INTO outbox (event_id, event_type, entity_type, entity_id, workspace_id, project_id, payload, created_at, initiator_id, initiator_type) VALUES ( @@ -414,7 +409,7 @@ class IssueAssigneeProxy(IssueAssignee): NEW.issue_id, NEW.workspace_id, NEW.project_id, - jsonb_build_object('data', enriched_data, 'previous_attributes', issue_data), + jsonb_build_object('data', enriched_data, 'previous_attributes', jsonb_build_object('assignee_ids', previous_assignee_ids)), now(), NEW.created_by_id, COALESCE(current_setting('plane.initiator_type', true), 'USER') @@ -430,11 +425,10 @@ class IssueAssigneeProxy(IssueAssignee): """, condition=None, ), - # Since we do a soft delete, we need to update the outbox when the assignee is removed pgtrigger.Trigger( name="issue_assignee_outbox_update", operation=pgtrigger.Update, - when=pgtrigger.After, + when=pgtrigger.Before, # Changed from After to Before func=""" DECLARE event_type_name TEXT; @@ -443,82 +437,79 @@ class IssueAssigneeProxy(IssueAssignee): previous_assignee_ids JSONB; label_ids JSONB; issue_data JSONB; - previous_issue_data JSONB; BEGIN - -- Determine event type based on issue type - SELECT - CASE - WHEN i.type_id IS NULL THEN 'issue.assignee.removed' - WHEN it.is_epic = true THEN 'epic.assignee.removed' - ELSE 'issue.assignee.removed' - END - INTO event_type_name - FROM issues i - LEFT JOIN issue_types it ON it.id = i.type_id - WHERE i.id = OLD.issue_id; - - -- If no issue found, default to 'issue.assignee.removed' - IF event_type_name IS NULL THEN - event_type_name := 'issue.assignee.removed'; + -- Only trigger on soft delete (deleted_at changing from NULL to NOT NULL) + IF OLD.deleted_at IS NULL AND NEW.deleted_at IS NOT NULL THEN + -- Determine event type based on issue type + SELECT + CASE + WHEN i.type_id IS NULL THEN 'issue.assignee.removed' + WHEN it.is_epic = true THEN 'epic.assignee.removed' + ELSE 'issue.assignee.removed' + END + INTO event_type_name + FROM issues i + LEFT JOIN issue_types it ON it.id = i.type_id + WHERE i.id = OLD.issue_id; + + -- If no issue found, default to 'issue.assignee.removed' + IF event_type_name IS NULL THEN + event_type_name := 'issue.assignee.removed'; + END IF; + + -- Get the complete issue data (excluding description fields) + SELECT to_jsonb(i) - 'description_html' - 'description_binary' - 'description' - 'description_stripped' + INTO issue_data + FROM issues i + WHERE i.id = OLD.issue_id; + + -- Fetch current assignee IDs (excluding the one being removed) + SELECT COALESCE(jsonb_agg(ia.assignee_id ORDER BY ia.created_at), '[]'::jsonb) + INTO current_assignee_ids + FROM issue_assignees ia + WHERE ia.issue_id = OLD.issue_id AND ia.deleted_at IS NULL AND ia.id != OLD.id; + + -- Fetch ALL assignee IDs (including the one being removed) for previous_attributes + SELECT COALESCE(jsonb_agg(ia.assignee_id ORDER BY ia.created_at), '[]'::jsonb) + INTO previous_assignee_ids + FROM issue_assignees ia + WHERE ia.issue_id = OLD.issue_id AND ia.deleted_at IS NULL; + + -- Fetch label IDs + SELECT COALESCE(jsonb_agg(il.label_id ORDER BY il.created_at), '[]'::jsonb) + INTO label_ids + FROM issue_labels il + WHERE il.issue_id = OLD.issue_id AND il.deleted_at IS NULL; + + -- Create enriched data with complete issue and current assignee IDs (after removal) + enriched_data := issue_data || jsonb_build_object( + 'assignee_ids', current_assignee_ids, + 'label_ids', label_ids + ); + + -- Create previous attributes with only the previous assignee IDs + BEGIN + INSERT INTO outbox (event_id, event_type, entity_type, entity_id, workspace_id, project_id, payload, created_at, initiator_id, initiator_type) + VALUES ( + gen_random_uuid(), + event_type_name, + 'issue_assignee', + OLD.issue_id, + OLD.workspace_id, + OLD.project_id, + jsonb_build_object('data', enriched_data, 'previous_attributes', jsonb_build_object('assignee_ids', previous_assignee_ids)), + now(), + NEW.updated_by_id, + COALESCE(current_setting('plane.initiator_type', true), 'USER') + ) + ON CONFLICT DO NOTHING; + EXCEPTION + WHEN others THEN + RAISE WARNING 'Outbox delete-event failed for issue_assignee %, reason: %', + OLD.issue_id, SQLERRM; + END; END IF; - - -- Get the complete issue data (excluding description fields) - SELECT to_jsonb(i) - 'description_html' - 'description_binary' - 'description' - 'description_stripped' - INTO issue_data - FROM issues i - WHERE i.id = OLD.issue_id; - - -- Fetch current assignee IDs (excluding the one being removed) - SELECT COALESCE(jsonb_agg(ia.assignee_id ORDER BY ia.created_at), '[]'::jsonb) - INTO current_assignee_ids - FROM issue_assignees ia - WHERE ia.issue_id = OLD.issue_id AND ia.deleted_at IS NULL AND ia.id != OLD.id; - - -- Fetch ALL assignee IDs (including the one being removed) for previous_attributes - SELECT COALESCE(jsonb_agg(ia.assignee_id ORDER BY ia.created_at), '[]'::jsonb) - INTO previous_assignee_ids - FROM issue_assignees ia - WHERE ia.issue_id = OLD.issue_id AND ia.deleted_at IS NULL; - - -- Fetch label IDs - SELECT COALESCE(jsonb_agg(il.label_id ORDER BY il.created_at), '[]'::jsonb) - INTO label_ids - FROM issue_labels il - WHERE il.issue_id = OLD.issue_id AND il.deleted_at IS NULL; - - -- Create enriched data with complete issue and current assignee IDs (after removal) - enriched_data := issue_data || jsonb_build_object( - 'assignee_ids', current_assignee_ids, - 'label_ids', label_ids - ); - - -- Create previous attributes with complete issue and ALL assignee IDs (including the removed one) - previous_issue_data := issue_data || jsonb_build_object( - 'assignee_ids', previous_assignee_ids, - 'label_ids', label_ids - ); - - BEGIN - INSERT INTO outbox (event_id, event_type, entity_type, entity_id, workspace_id, project_id, payload, created_at, initiator_id, initiator_type) - VALUES ( - gen_random_uuid(), - event_type_name, - 'issue_assignee', - OLD.issue_id, - OLD.workspace_id, - OLD.project_id, - jsonb_build_object('data', enriched_data, 'previous_attributes', previous_issue_data), - now(), - NEW.updated_by_id, - COALESCE(current_setting('plane.initiator_type', true), 'USER') - ) - ON CONFLICT DO NOTHING; - EXCEPTION - WHEN others THEN - RAISE WARNING 'Outbox delete-event failed for issue_assignee %, reason: %', - OLD.issue_id, SQLERRM; - END; - RETURN OLD; + RETURN NEW; END; """, condition=None, @@ -590,12 +581,7 @@ class IssueLabelProxy(IssueLabel): 'label_ids', label_ids ); - -- Create previous attributes with label IDs that existed before the addition - issue_data := issue_data || jsonb_build_object( - 'assignee_ids', assignee_ids, - 'label_ids', previous_label_ids - ); - + -- Create previous attributes with only the previous label IDs BEGIN INSERT INTO outbox (event_id, event_type, entity_type, entity_id, workspace_id, project_id, payload, created_at, initiator_id, initiator_type) VALUES ( @@ -605,7 +591,7 @@ class IssueLabelProxy(IssueLabel): NEW.issue_id, NEW.workspace_id, NEW.project_id, - jsonb_build_object('data', enriched_data, 'previous_attributes', issue_data), + jsonb_build_object('data', enriched_data, 'previous_attributes', jsonb_build_object('label_ids', previous_label_ids)), now(), NEW.created_by_id, COALESCE(current_setting('plane.initiator_type', true), 'USER') @@ -621,11 +607,10 @@ class IssueLabelProxy(IssueLabel): """, condition=None, ), - # Since we do a soft delete, we need to update the outbox when the label is removed pgtrigger.Trigger( name="issue_label_outbox_update", operation=pgtrigger.Update, - when=pgtrigger.After, + when=pgtrigger.Before, # Changed from After to Before func=""" DECLARE event_type_name TEXT; @@ -634,82 +619,79 @@ class IssueLabelProxy(IssueLabel): current_label_ids JSONB; previous_label_ids JSONB; issue_data JSONB; - previous_issue_data JSONB; BEGIN - -- Determine event type based on issue type - SELECT - CASE - WHEN i.type_id IS NULL THEN 'issue.label.removed' - WHEN it.is_epic = true THEN 'epic.label.removed' - ELSE 'issue.label.removed' - END - INTO event_type_name - FROM issues i - LEFT JOIN issue_types it ON it.id = i.type_id - WHERE i.id = OLD.issue_id; - - -- If no issue found, default to 'issue.label.removed' - IF event_type_name IS NULL THEN - event_type_name := 'issue.label.removed'; - END IF; - - -- Get the complete issue data (excluding description fields) - SELECT to_jsonb(i) - 'description_html' - 'description_binary' - 'description' - 'description_stripped' - INTO issue_data - FROM issues i - WHERE i.id = OLD.issue_id; - - -- Fetch assignee IDs - SELECT COALESCE(jsonb_agg(ia.assignee_id ORDER BY ia.created_at), '[]'::jsonb) - INTO assignee_ids - FROM issue_assignees ia - WHERE ia.issue_id = OLD.issue_id AND ia.deleted_at IS NULL; - - -- Fetch current label IDs (excluding the one being removed) - SELECT COALESCE(jsonb_agg(il.label_id ORDER BY il.created_at), '[]'::jsonb) - INTO current_label_ids - FROM issue_labels il - WHERE il.issue_id = OLD.issue_id AND il.deleted_at IS NULL AND il.id != OLD.id; - - -- Fetch ALL label IDs (including the one being removed) for previous_attributes - SELECT COALESCE(jsonb_agg(il.label_id ORDER BY il.created_at), '[]'::jsonb) - INTO previous_label_ids - FROM issue_labels il - WHERE il.issue_id = OLD.issue_id AND il.deleted_at IS NULL; - - -- Create enriched data with complete issue, assignee IDs, and current label IDs (after removal) - enriched_data := issue_data || jsonb_build_object( - 'assignee_ids', assignee_ids, - 'label_ids', current_label_ids - ); - - -- Create previous attributes with complete issue, assignee IDs, and ALL label IDs (including the removed one) - previous_issue_data := issue_data || jsonb_build_object( - 'assignee_ids', assignee_ids, - 'label_ids', previous_label_ids - ); - - BEGIN - INSERT INTO outbox (event_id, event_type, entity_type, entity_id, workspace_id, project_id, payload, created_at, initiator_id, initiator_type) - VALUES ( - gen_random_uuid(), - event_type_name, - 'issue_label', - OLD.issue_id, - OLD.workspace_id, - OLD.project_id, - jsonb_build_object('data', enriched_data, 'previous_attributes', previous_issue_data), - now(), - NEW.updated_by_id, - COALESCE(current_setting('plane.initiator_type', true), 'USER') - ) - ON CONFLICT DO NOTHING; - EXCEPTION - WHEN others THEN - RAISE WARNING 'Outbox delete-event failed for issue_label %, reason: %', + -- Only trigger on soft delete (deleted_at changing from NULL to NOT NULL) + IF OLD.deleted_at IS NULL AND NEW.deleted_at IS NOT NULL THEN + -- Determine event type based on issue type + SELECT + CASE + WHEN i.type_id IS NULL THEN 'issue.label.removed' + WHEN it.is_epic = true THEN 'epic.label.removed' + ELSE 'issue.label.removed' + END + INTO event_type_name + FROM issues i + LEFT JOIN issue_types it ON it.id = i.type_id + WHERE i.id = OLD.issue_id; + + -- If no issue found, default to 'issue.label.removed' + IF event_type_name IS NULL THEN + event_type_name := 'issue.label.removed'; + END IF; + + -- Get the complete issue data (excluding description fields) + SELECT to_jsonb(i) - 'description_html' - 'description_binary' - 'description' - 'description_stripped' + INTO issue_data + FROM issues i + WHERE i.id = OLD.issue_id; + + -- Fetch assignee IDs + SELECT COALESCE(jsonb_agg(ia.assignee_id ORDER BY ia.created_at), '[]'::jsonb) + INTO assignee_ids + FROM issue_assignees ia + WHERE ia.issue_id = OLD.issue_id AND ia.deleted_at IS NULL; + + -- Fetch current label IDs (excluding the one being removed) + SELECT COALESCE(jsonb_agg(il.label_id ORDER BY il.created_at), '[]'::jsonb) + INTO current_label_ids + FROM issue_labels il + WHERE il.issue_id = OLD.issue_id AND il.deleted_at IS NULL AND il.id != OLD.id; + + -- Fetch ALL label IDs (including the one being removed) for previous_attributes + SELECT COALESCE(jsonb_agg(il.label_id ORDER BY il.created_at), '[]'::jsonb) + INTO previous_label_ids + FROM issue_labels il + WHERE il.issue_id = OLD.issue_id AND il.deleted_at IS NULL; + + -- Create enriched data with complete issue, assignee IDs, and current label IDs (after removal) + enriched_data := issue_data || jsonb_build_object( + 'assignee_ids', assignee_ids, + 'label_ids', current_label_ids + ); + + -- Create previous attributes with only the previous label IDs (including the removed one) + BEGIN + INSERT INTO outbox (event_id, event_type, entity_type, entity_id, workspace_id, project_id, payload, created_at, initiator_id, initiator_type) + VALUES ( + gen_random_uuid(), + event_type_name, + 'issue_label', + OLD.issue_id, + OLD.workspace_id, + OLD.project_id, + jsonb_build_object('data', enriched_data, 'previous_attributes', jsonb_build_object('label_ids', previous_label_ids)), + now(), + NEW.updated_by_id, + COALESCE(current_setting('plane.initiator_type', true), 'USER') + ) + ON CONFLICT DO NOTHING; + EXCEPTION + WHEN others THEN + RAISE WARNING 'Outbox delete-event failed for issue_label %, reason: %', OLD.issue_id, SQLERRM; - END; - RETURN OLD; + END; + END IF; + RETURN NEW; END; """, condition=None, diff --git a/apps/api/plane/settings/common.py b/apps/api/plane/settings/common.py index ead345f3c4..179f0cf251 100644 --- a/apps/api/plane/settings/common.py +++ b/apps/api/plane/settings/common.py @@ -323,6 +323,8 @@ CELERY_IMPORTS = ( "plane.ee.bgtasks.recurring_work_item_task", # silo tasks "plane.silo.bgtasks.integration_apps_task", + # event stream tasks + "plane.event_stream.bgtasks.outbox_cleaner", ) # Application Envs diff --git a/apps/api/plane/silo/bgtasks/bulk_update_issue_relations_task.py b/apps/api/plane/silo/bgtasks/bulk_update_issue_relations_task.py index 58eae522f0..90da06c292 100644 --- a/apps/api/plane/silo/bgtasks/bulk_update_issue_relations_task.py +++ b/apps/api/plane/silo/bgtasks/bulk_update_issue_relations_task.py @@ -73,7 +73,9 @@ def bulk_update_issue_relations_task(job_id: str, user_id: str | None = None): ) with transaction.atomic(): with connection.cursor() as cur: - cur.execute("SET LOCAL plane.initiator_type = 'SYSTEM.IMPORT'") + cur.execute( + "SELECT set_config('plane.initiator_type', 'SYSTEM.IMPORT', true)" + ) # Bulk create relations, ignoring any duplicates IssueRelation.objects.bulk_create(issue_relations, ignore_conflicts=True) @@ -87,7 +89,9 @@ def bulk_update_issue_relations_task(job_id: str, user_id: str | None = None): continue with connection.cursor() as cur: - cur.execute("SET LOCAL plane.initiator_type = 'SYSTEM.IMPORT'") + cur.execute( + "SELECT set_config('plane.initiator_type', 'SYSTEM.IMPORT', true)" + ) # Update parent_id Issue.objects.filter(id=issue_id).update( parent_id=parent_id, updated_by_id=user_id