Function enqueue_membership_refresh

api/segment_membership/services.py:41–76

Queue a per-project segment membership count refresh. Pass `delay_until` to schedule the refresh for a future time, e.g. to let an Edge CDC tombstone land in ClickHouse before recounting. No-op when the org has the feature off, or when a refresh for the project is already pending.

(
    project: Project,
    *,
    delay_until: datetime | None = None,
)

def enqueue_membership_refresh(
    project: Project,
    *,
    delay_until: datetime | None = None,
) -> None:
    """Queue a per-project segment membership count refresh.

    Pass `delay_until` to schedule the refresh for a future time, e.g. to let an
    Edge CDC tombstone land in ClickHouse before recounting.

    No-op when the org has the feature off, or when a refresh for the project is
    already pending. A pending refresh scheduled sooner than `delay_until` is
    pushed out to it.
    """
    if not is_membership_enabled(project.organisation):
        return

    from segment_membership.tasks import refresh_project_segment_counts

    pending = Task.objects.filter(
        task_identifier=refresh_project_segment_counts.task_identifier,
        completed=False,
        num_failures__lt=3,
        serialized_args=Task.serialize_data((project.id,)),
    )
    if pending.exists():
        if delay_until is not None:
            pending.filter(
                Q(scheduled_for__isnull=True) | Q(scheduled_for__lt=delay_until),
            ).update(scheduled_for=delay_until)
        return

    refresh_project_segment_counts.delay(
        args=(project.id,),
        delay_until=delay_until,
    )


@contextmanager