Mirrors/plane
1
0
Fork 0
mirror of https://github.com/makeplane/plane.git synced 2026-07-13 14:01:45 +02:00
Code Issues Packages Projects Releases Wiki Activity

[WEB-4667] chore: outbox event payload and change the way transaction variable is set (#3888)

Browse Source
This commit is contained in:
Nikhil
2025-08-12 19:49:51 +05:30
committed by GitHub
parent 3a0b6e39c3
commit ccf24a6525
6 changed files with 247 additions and 170 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
apps/api/plane/bgtasks/data_import_task.py
View File

@@ -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(

2
apps/api/plane/event_stream/bgtasks/outbox_cleaner.py
View File

@@ -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)"

87
apps/api/plane/event_stream/migrations/0003_remove_issueassigneeproxy_issue_assignee_outbox_insert_and_more.py Normal file
View File

@@ -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",
),
),
),
]

314
apps/api/plane/event_stream/models/issue.py
View File

@@ -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,

2
apps/api/plane/settings/common.py
View File

@@ -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

8
apps/api/plane/silo/bgtasks/bulk_update_issue_relations_task.py
View File

@@ -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