res_stasis_broadcast.c

Go to the documentation of this file.

/*
 * Asterisk -- An open source telephony toolkit.
 *
 * Copyright (C) 2026, Aurora Innovation AB
 *
 * Daniel Donoghue <daniel.donoghue@aurorainnovation.com>
 *
 * See http://www.asterisk.org for more information about
 * the Asterisk project. Please do not directly contact
 * any of the maintainers of this project for assistance;
 * the project provides a web site, mailing lists and IRC
 * channels for your use.
 *
 * This program is free software, distributed under the terms of
 * the GNU General Public License Version 2. See the LICENSE file
 * at the top of the source tree.
 */
 
/*! \file
 *
 * \brief Stasis application broadcast resource
 *
 * \author Daniel Donoghue <daniel.donoghue@aurorainnovation.com>
 */
 
/*** MODULEINFO
    <depend type="module">res_stasis</depend>
    <depend type="module">res_ari</depend>
    <support_level>extended</support_level>
 ***/
 
#include "asterisk.h"
 
#include <errno.h>
#include <regex.h>
 
#include "asterisk/astobj2.h"
#include "asterisk/channel.h"
#include "asterisk/http.h"
#include "asterisk/json.h"
#include "asterisk/lock.h"
#include "asterisk/module.h"
#include "asterisk/pbx.h"
#include "asterisk/stasis_app.h"
#include "asterisk/stasis_app_impl.h"
#include "asterisk/stasis_channels.h"
#include "asterisk/time.h"
#include "asterisk/utils.h"
 
#define AST_API_MODULE  /* Mark this as the module providing the API */
#include "asterisk/stasis_app_broadcast.h"
 
#define BROADCAST_BUCKETS 37
 
/*! \brief Maximum length for app_filter regex pattern */
#define MAX_REGEX_LENGTH 256
 
/*! \brief Maximum depth for regex group nesting */
#define MAX_GROUP_DEPTH 10
 
/*! \brief Maximum number of nested quantifiers in regex */
#define MAX_NESTED_QUANTIFIERS 3
 
/*! \brief Maximum value for brace quantifier bounds {m,n} */
#define MAX_QUANTIFIER_BOUND 100
 
/*! \brief Maximum alternations allowed in deeply nested groups */
#define MAX_ALTERNATIONS 20
 
/*! \brief Group depth threshold for alternation limits */
#define ALTERNATION_DEPTH_THRESHOLD 2
 
/*! \brief Maximum broadcast timeout in milliseconds (24 hours) */
#define MAX_BROADCAST_TIMEOUT_MS (24 * 60 * 60 * 1000)
 
/*! \brief Interval in ms between hangup checks while waiting for a claim */
#define BROADCAST_POLL_INTERVAL_MS 200
 
/*! \brief Broadcast context stored on channel */
struct stasis_broadcast_ctx {
    /*! The unique ID of the channel */
    char channel_id[AST_MAX_PUBLIC_UNIQUEID];
    /*! Name of the winning application (dynamically allocated, NULL until claimed) */
    char *winner_app;
    /*! Regex pattern used to filter broadcast recipients */
    char app_filter[MAX_REGEX_LENGTH + 1];
    /*! Compiled regex for app_filter (valid only when filter_compiled is set) */
    regex_t compiled_filter;
    /*! Flag indicating if channel was claimed */
    unsigned int claimed:1;
    /*! Whether compiled_filter is valid and must be freed */
    unsigned int filter_compiled:1;
    /*! Set when the PBX thread retrieves the winner; prevents late claims */
    unsigned int finished:1;
    /*! Broadcast behaviour flags (STASIS_BROADCAST_FLAG_*) */
    unsigned int flags;
    /*! Reference to the global container (prevents use-after-free during module unload) */
    struct ao2_container *container;
    /*! Condition variable for claim notification */
    ast_cond_t cond;
};
 
/*! \brief Container for all active broadcast contexts */
static struct ao2_container *broadcast_contexts;
 
/*! \brief Destructor for broadcast datastore
 *
 * Called when the channel is destroyed. Ensures the broadcast context
 * is unlinked from the global container even if the caller never
 * reached stasis_app_broadcast_cleanup (e.g. abnormal channel teardown).
 */
static void broadcast_datastore_destroy(void *data)
{
    struct stasis_broadcast_ctx *ctx = data;
 
    if (ctx->container) {
        ao2_unlink(ctx->container, ctx);
    }
    ao2_cleanup(ctx);
}
 
/*! \brief Datastore information for broadcast context */
static const struct ast_datastore_info broadcast_datastore_info = {
    .type = "stasis_broadcast_context",
    .destroy = broadcast_datastore_destroy,
};
 
AO2_STRING_FIELD_HASH_FN(stasis_broadcast_ctx, channel_id)
AO2_STRING_FIELD_CMP_FN(stasis_broadcast_ctx, channel_id)
 
/*! \brief Destructor for broadcast context */
static void broadcast_ctx_destructor(void *obj)
{
    struct stasis_broadcast_ctx *ctx = obj;
    ast_free(ctx->winner_app);
    if (ctx->filter_compiled) {
        regfree(&ctx->compiled_filter);
    }
    ao2_cleanup(ctx->container);
    ast_cond_destroy(&ctx->cond);
}
 
static int validate_regex_pattern(const char *pattern);
 
/*! \brief Create a new broadcast context
 *
 * Validates and compiles the app_filter regex if provided. On regex
 * failure the context is still created but broadcasts will be sent
 * to all applications (i.e. no filtering).
 */
static struct stasis_broadcast_ctx *broadcast_ctx_create(
    const char *channel_id, const char *app_filter, unsigned int flags)
{
    struct stasis_broadcast_ctx *ctx;
 
    ctx = ao2_alloc(sizeof(*ctx), broadcast_ctx_destructor);
    if (!ctx) {
        return NULL;
    }
 
    /* ao2_alloc zeroes the struct; only set non-zero fields explicitly */
    ast_copy_string(ctx->channel_id, channel_id, sizeof(ctx->channel_id));
    ctx->flags = flags;
    ctx->container = ao2_bump(broadcast_contexts);
    ast_cond_init(&ctx->cond, NULL);
 
    /* Validate and compile app_filter regex if provided */
    if (!ast_strlen_zero(app_filter)) {
        ast_copy_string(ctx->app_filter, app_filter, sizeof(ctx->app_filter));
        if (validate_regex_pattern(app_filter) != 0) {
            ast_log(LOG_WARNING,
                "Channel %s: rejecting app_filter regex as potentially dangerous: %s\n",
                channel_id, app_filter);
        } else if (regcomp(&ctx->compiled_filter, app_filter,
                REG_EXTENDED | REG_NOSUB) != 0) {
            ast_log(LOG_WARNING,
                "Channel %s: failed to compile app_filter regex '%s'\n",
                channel_id, app_filter);
        } else {
            ctx->filter_compiled = 1;
        }
 
        if (!ctx->filter_compiled) {
            ast_log(LOG_WARNING,
                "Channel %s: proceeding without application filtering due to invalid regex\n",
                channel_id);
        }
    }
 
    ast_debug(1, "Created broadcast context for channel %s (filter: %s, flags: 0x%x)\n",
        ctx->channel_id,
        ctx->filter_compiled ? ctx->app_filter : "none",
        ctx->flags);
 
    return ctx;
}
 
/*!
 * \brief Validate a regex pattern for safety
 *
 * Checks that the regex pattern is within length limits and doesn't contain
 * patterns that could cause excessive backtracking or denial of service.
 *
 * \param pattern The regex pattern to validate
 * \return 0 if valid, -1 if invalid
 */
static int validate_regex_pattern(const char *pattern)
{
    size_t len;
    int group_depth = 0;
    int quantified_groups = 0;
    int in_class = 0; /* Inside [...] */
    /* Track alternations per group depth. Index 0 is outside groups and unused. */
    int alternations_per_depth[MAX_GROUP_DEPTH + 1] = { 0 };
    const char *p;
 
    if (ast_strlen_zero(pattern)) {
        return 0; /* Empty pattern is valid (will be skipped) */
    }
 
    /* Check maximum length to prevent excessive regex compilation time */
    len = strlen(pattern);
    if (len > MAX_REGEX_LENGTH) {
        ast_debug(3, "Regex pattern exceeds maximum length of %d characters (got %zu)\n",
            MAX_REGEX_LENGTH, len);
        return -1;
    }
 
    /*
     * Check for potentially dangerous patterns that could cause
     * excessive regex compilation or matching time. Look for:
     * - Excessive group nesting depth
     * - Too many quantified groups (groups followed by +, *, or ?)
     *
     * Note: This is a heuristic approach that catches common dangerous
     * patterns. Combined with the length limit, it provides reasonable
     * protection against ReDoS while allowing legitimate regex usage.
     */
    for (p = pattern; *p; p++) {
        /* Handle character classes: enter on unescaped '[' and exit on unescaped ']' */
        if (!in_class && *p == '[' && (p == pattern || *(p - 1) != '\\')) {
            in_class = 1;
            /* In POSIX ERE, ']' immediately after '[' or '[^' is a
             * literal, not the end of the class.  Advance past the
             * optional negation caret and the literal ']' so the
             * main loop does not leave in_class prematurely. */
            if (*(p + 1) == '^') {
                p++;
            }
            if (*(p + 1) == ']') {
                p++;
            }
            continue;
        } else if (in_class) {
            if (*p == '\\') {
                /* Skip the next escaped character inside character class */
                if (*(p + 1)) {
                    p++;
                }
                continue;
            }
            if (*p == ']') {
                in_class = 0;
            }
            /* Ignore everything inside character classes for heuristics */
            continue;
        }
        switch (*p) {
        case '(':
            group_depth++;
            if (group_depth > MAX_GROUP_DEPTH) {
                ast_debug(3, "Regex pattern has too many nested groups (max %d)\n",
                    MAX_GROUP_DEPTH);
                return -1;
            }
            /* Reset alternation counter for newly entered group depth */
            alternations_per_depth[group_depth] = 0;
            break;
        case ')':
            if (group_depth > 0) {
                /* Clear alternations count for this depth before leaving */
                alternations_per_depth[group_depth] = 0;
                group_depth--;
            }
            break;
        case '+':
        case '*':
        case '?':
            /*
             * Count quantified groups - patterns like (...)+ or (...)*
             * Too many of these can cause slow matching on certain inputs.
             */
            if (p > pattern && *(p - 1) == ')') {
                quantified_groups++;
            }
            break;
        case '{': {
            /* Parse POSIX quantifier {m}, {m,}, {m,n} with overflow and bound checks */
            const char *q = p + 1;
            long m = 0, n = -1; /* n=-1 means open upper bound */
            int valid = 0;
            int digit;
            int overflow = 0;
 
            if (*q >= '0' && *q <= '9') {
                /* Parse m safely */
                while (*q >= '0' && *q <= '9') {
                    digit = (*q - '0');
                    if (m > (LONG_MAX - digit) / 10) { /* overflow on next step */
                        overflow = 1;
                        break;
                    }
                    m = (m * 10) + digit;
                    if (m > MAX_QUANTIFIER_BOUND) { /* early bound exceed */
                        overflow = 1;
                        break;
                    }
                    q++;
                }
                if (!overflow && *q == ',') {
                    q++;
                    if (*q >= '0' && *q <= '9') {
                        long nn = 0;
                        while (*q >= '0' && *q <= '9') {
                            digit = (*q - '0');
                            if (nn > (LONG_MAX - digit) / 10) {
                                overflow = 1;
                                break;
                            }
                            nn = (nn * 10) + digit;
                            if (nn > MAX_QUANTIFIER_BOUND) {
                                overflow = 1;
                                break;
                            }
                            q++;
                        }
                        n = nn;
                    } else {
                        n = -1; /* open upper bound */
                    }
                } else if (!overflow) {
                    n = m; /* {m} */
                }
                if (!overflow && *q == '}') {
                    valid = 1;
                }
            }
            if (overflow) {
                ast_debug(3, "Regex quantifier overflow or exceeds max bound (max %d)\n", MAX_QUANTIFIER_BOUND);
                return -1;
            }
            if (valid) {
                /* Additional bounds check (defensive) */
                if (m > MAX_QUANTIFIER_BOUND || (n != -1 && n > MAX_QUANTIFIER_BOUND)) {
                    ast_debug(3, "Regex quantifier bounds too large (max %d)\n", MAX_QUANTIFIER_BOUND);
                    return -1;
                }
                if (p > pattern && *(p - 1) == ')') {
                    quantified_groups++;
                }
                p = q; /* q currently points to '}' */
            }
            break;
        }
        case '|':
            if (group_depth > 0) {
                alternations_per_depth[group_depth]++;
                if (group_depth > ALTERNATION_DEPTH_THRESHOLD &&
                    alternations_per_depth[group_depth] > MAX_ALTERNATIONS) {
                    ast_debug(3,
                        "Regex has too many alternations in deep group (depth %d, count %d, max %d)\n",
                        group_depth,
                        alternations_per_depth[group_depth],
                        MAX_ALTERNATIONS);
                    return -1;
                }
            }
            break;
        case '\\':
            /*
             * Skip the next character entirely from heuristic processing.
             * This ensures escaped characters (metacharacters in BRE or literals
             * in ERE like \‍(, \‍), \+, \*, \?, etc.) do not affect group depth
             * or quantified group counts.
             */
            if (*(p + 1)) {
                p++;
            }
            /* Continue to next loop iteration without evaluating the escaped char */
            continue;
        }
    }
 
    /*
     * Reject patterns with too many quantified groups, as these are
     * often indicators of potentially slow patterns that could be
     * exploited for denial of service.
     */
    if (quantified_groups > MAX_NESTED_QUANTIFIERS) {
        ast_debug(3, "Regex pattern has too many quantified groups (max %d)\n",
            MAX_NESTED_QUANTIFIERS);
        return -1;
    }
 
    return 0;
}
 
/*! \brief Create and send broadcast event to all applications
 *
 * Uses the compiled regex cached in \a ctx for application filtering.
 */
static int send_broadcast_event(struct ast_channel *chan,
    struct stasis_broadcast_ctx *ctx)
{
    RAII_VAR(struct ast_json *, event, NULL, ast_json_unref);
    RAII_VAR(struct ast_channel_snapshot *, snapshot, NULL, ao2_cleanup);
    RAII_VAR(struct ao2_container *, apps, NULL, ao2_cleanup);
    struct ao2_iterator iter;
    char *app_name;
    const char *caller = NULL;
    const char *called = NULL;
 
    /* Get snapshot and caller/called info under a single channel lock */
    ast_channel_lock(chan);
    snapshot = ao2_bump(ast_channel_snapshot(chan));
    caller = ast_strdupa(S_OR(ast_channel_caller(chan)->id.number.str, ""));
    called = ast_strdupa(S_OR(ast_channel_exten(chan), ""));
    ast_channel_unlock(chan);
 
    /* Build the broadcast event.  Channel variables configured in
     * ari.conf "channelvars" are already included in the channel
     * snapshot produced by ast_channel_snapshot_to_json(). */
    event = ast_json_pack("{s: s, s: o, s: o, s: s?, s: s?}",
        "type", "CallBroadcast",
        "timestamp", ast_json_timeval(ast_tvnow(), NULL),
        "channel", ast_channel_snapshot_to_json(snapshot, NULL),
        "caller", caller,
        "called", called);
 
    if (!event) {
        ast_log(LOG_ERROR, "Channel %s: failed to create broadcast event\n",
            ast_channel_name(chan));
        return -1;
    }
 
    /* Get all registered applications */
    apps = stasis_app_get_all();
    if (!apps) {
        ast_log(LOG_ERROR, "Channel %s: failed to get stasis applications\n",
            ast_channel_name(chan));
        return -1;
    }
 
    ast_debug(2, "Broadcasting to %d registered Stasis applications\n",
        ao2_container_count(apps));
 
    /*
     * Broadcast to all matching applications.
     *
     * We collect matching apps into a plain array, Fisher-Yates shuffle it,
     * then call stasis_app_send() for each.  stasis_app_send() writes
     * directly to each app's WebSocket socket synchronously on the calling
     * thread.  The shuffle ensures no single ARI application is consistently
     * first to receive the event — every app gets a fair chance to claim the
     * channel regardless of its position in the ao2 hash container.
     */
    {
        int app_count;
        char **matching_arr;
        int n = 0;
        int i;
 
        app_count = ao2_container_count(apps);
        if (app_count == 0) {
            ast_debug(2, "Channel %s: no Stasis applications registered\n",
                ast_channel_uniqueid(chan));
            return 0;
        }
 
        matching_arr = ast_malloc(app_count * sizeof(*matching_arr));
        if (!matching_arr) {
            ast_log(LOG_ERROR, "Channel %s: failed to allocate matching apps array\n",
                ast_channel_name(chan));
            return -1;
        }
 
        /* First pass: collect all matching app names (transfer refs to array) */
        iter = ao2_iterator_init(apps, 0);
        while ((app_name = ao2_iterator_next(&iter)) && n < app_count) {
            if (ctx->filter_compiled &&
                regexec(&ctx->compiled_filter, app_name, 0, NULL, 0) == REG_NOMATCH) {
                ast_debug(3, "App '%s' does not match filter, skipping\n", app_name);
                ao2_ref(app_name, -1);
                continue;
            }
            matching_arr[n++] = app_name; /* ref transferred to array */
        }
        ao2_iterator_destroy(&iter);
 
        ast_debug(2, "Broadcasting channel %s to %d matching applications\n",
            ast_channel_uniqueid(chan), n);
 
        /* Fisher-Yates shuffle: randomise delivery order so no app is
         * consistently first to receive the broadcast event. */
        for (i = n - 1; i > 0; i--) {
            int j = ast_random() % (i + 1);
            char *tmp = matching_arr[i];
            matching_arr[i] = matching_arr[j];
            matching_arr[j] = tmp;
        }
 
        /*
         * Second pass: send to each matching app.  A deep copy of the event
         * is required for each call because stasis_app_send() mutates the
         * message in-place (adds "asterisk_id" via ast_json_object_set).
         */
        for (i = 0; i < n; i++) {
            char *match_name = matching_arr[i];
            struct ast_json *event_copy;
 
            ast_debug(3, "Sending broadcast to app '%s'\n", match_name);
 
            event_copy = ast_json_deep_copy(event);
            if (!event_copy) {
                ast_log(LOG_ERROR,
                    "Channel %s: failed to deep-copy event for app '%s'\n",
                    ast_channel_uniqueid(chan), match_name);
                ao2_ref(match_name, -1);
                continue;
            }
 
            stasis_app_send(match_name, event_copy);
            ast_json_unref(event_copy);
            ao2_ref(match_name, -1);
        }
 
        ast_free(matching_arr);
    }
 
    return 0;
}
 
/*!
 * \brief Start a broadcast for a channel
 * \param chan The channel to broadcast
 * \param timeout_ms Timeout in milliseconds
 * \param app_filter Optional regex filter for applications
 * \return 0 on success, -1 on error
 */
int AST_OPTIONAL_API_NAME(stasis_app_broadcast_channel)(struct ast_channel *chan, int timeout_ms,
    const char *app_filter, unsigned int flags)
{
    RAII_VAR(struct stasis_broadcast_ctx *, ctx, NULL, ao2_cleanup);
    struct ast_datastore *datastore;
 
    if (!chan) {
        return -1;
    }
 
    if (!broadcast_contexts) {
        return -1;
    }
 
    /* Remove any previous broadcast datastore from a prior attempt.
     * This supports failover scenarios where StasisBroadcast() is
     * called multiple times for the same channel.  The datastore
     * destructor unlinks the old context from the container. */
    {
        struct ast_datastore *old_ds;
        ast_channel_lock(chan);
        old_ds = ast_channel_datastore_find(chan, &broadcast_datastore_info, NULL);
        if (old_ds) {
            ast_channel_datastore_remove(chan, old_ds);
        }
        ast_channel_unlock(chan);
        if (old_ds) {
            ast_datastore_free(old_ds);
        }
    }
 
    /* Create broadcast context (validates and compiles app_filter regex) */
    ctx = broadcast_ctx_create(ast_channel_uniqueid(chan), app_filter, flags);
    if (!ctx) {
        ast_log(LOG_ERROR, "Channel %s: failed to create broadcast context\n",
            ast_channel_uniqueid(chan));
        return -1;
    }
 
    /* Store context in container */
    ao2_link(broadcast_contexts, ctx);
 
    /* Create and attach datastore to channel */
    datastore = ast_datastore_alloc(&broadcast_datastore_info, ast_channel_uniqueid(chan));
    if (!datastore) {
        ast_log(LOG_ERROR, "Channel %s: failed to allocate broadcast datastore\n",
            ast_channel_uniqueid(chan));
        ao2_unlink(broadcast_contexts, ctx);
        return -1;
    }
 
    datastore->data = ao2_bump(ctx);
    ast_channel_lock(chan);
    if (ast_channel_datastore_add(chan, datastore)) {
        ast_channel_unlock(chan);
        ast_log(LOG_ERROR, "Channel %s: failed to attach broadcast datastore\n",
            ast_channel_uniqueid(chan));
        ast_datastore_free(datastore);
        ao2_unlink(broadcast_contexts, ctx);
        return -1;
    }
    ast_channel_unlock(chan);
 
    ast_debug(1, "Starting broadcast for channel %s (timeout: %dms, filter: %s)\n",
        ast_channel_uniqueid(chan), timeout_ms, app_filter ? app_filter : "none");
 
    /* Send broadcast event to all matching applications */
    if (send_broadcast_event(chan, ctx) != 0) {
        ast_log(LOG_ERROR, "Channel %s: failed to send broadcast event\n",
            ast_channel_uniqueid(chan));
        ast_channel_lock(chan);
        ast_channel_datastore_remove(chan, datastore);
        ast_channel_unlock(chan);
        ast_datastore_free(datastore);
        ao2_unlink(broadcast_contexts, ctx);
        return -1;
    }
 
    return 0;
}
 
/*!
 * \brief Attempt to claim a broadcast channel
 * \param channel_id The unique ID of the channel
 * \param app_name The name of the application claiming the channel
 * \return 0 if claim successful, -1 if channel not found, -2 if already claimed
 */
int AST_OPTIONAL_API_NAME(stasis_app_claim_channel)(const char *channel_id, const char *app_name)
{
    RAII_VAR(struct stasis_broadcast_ctx *, ctx, NULL, ao2_cleanup);
 
    if (ast_strlen_zero(channel_id) || ast_strlen_zero(app_name)) {
        return -1;
    }
 
    if (!broadcast_contexts) {
        return -1;
    }
 
    /* Find broadcast context */
    ctx = ao2_find(broadcast_contexts, channel_id, OBJ_SEARCH_KEY);
    if (!ctx) {
        ast_debug(1, "No broadcast context found for channel %s\n", channel_id);
        return -1;
    }
 
    /* Atomically check and set claimed flag.
     * Check claimed before finished: if the channel was claimed and then the
     * broadcast finished, a late claim should return -2 (409 Conflict) rather
     * than -1 (404) so callers can distinguish "already taken" from "not found". */
    ao2_lock(ctx);
    if (ctx->claimed) {
        ast_debug(1, "Channel %s already claimed by %s (attempt by %s denied)\n",
            channel_id, ctx->winner_app ? ctx->winner_app : "(unknown)", app_name);
        ao2_unlock(ctx);
        return -2;
    }
    if (ctx->finished) {
        ast_debug(1, "Channel %s broadcast already finished (late claim by %s rejected)\n",
            channel_id, app_name);
        ao2_unlock(ctx);
        return -1;
    }
    ctx->winner_app = ast_strdup(app_name);
    if (!ctx->winner_app) {
        ast_log(LOG_ERROR,
            "Failed to allocate winner app name for channel %s\n",
            channel_id);
        ao2_unlock(ctx);
        return -1;
    }
    ctx->claimed = 1;
    ast_verb(3, "Channel %s claimed by application %s\n",
        channel_id, app_name);
    /* Signal waiting thread that channel was claimed */
    ast_cond_signal(&ctx->cond);
    ao2_unlock(ctx);
 
    /* Send CallClaimed event to matching apps */
    if (!(ctx->flags & STASIS_BROADCAST_FLAG_SUPPRESS_CLAIMED)) {
        RAII_VAR(struct ast_channel_snapshot *, snapshot, NULL, ao2_cleanup);
        RAII_VAR(struct ast_json *, event, NULL, ast_json_unref);
        RAII_VAR(struct ao2_container *, apps, NULL, ao2_cleanup);
        struct ao2_iterator iter;
        char *app_name_iter;
 
        snapshot = ast_channel_snapshot_get_latest(channel_id);
        if (snapshot) {
            event = ast_json_pack("{s: s, s: o, s: o, s: s}",
                "type", "CallClaimed",
                "timestamp", ast_json_timeval(ast_tvnow(), NULL),
                "channel", ast_channel_snapshot_to_json(snapshot, NULL),
                "winner_app", app_name);
        }
        if (event) {
            apps = stasis_app_get_all();
        }
        if (apps) {
            iter = ao2_iterator_init(apps, 0);
            while ((app_name_iter = ao2_iterator_next(&iter))) {
                struct ast_json *event_copy;
 
                /* Only send to apps that matched the original broadcast filter */
                if (ctx->filter_compiled &&
                    regexec(&ctx->compiled_filter, app_name_iter,
                        0, NULL, 0) == REG_NOMATCH) {
                    ao2_ref(app_name_iter, -1);
                    continue;
                }
 
                event_copy = ast_json_deep_copy(event);
                if (!event_copy) {
                    ao2_ref(app_name_iter, -1);
                    continue;
                }
 
                stasis_app_send(app_name_iter, event_copy);
                ast_json_unref(event_copy);
                ao2_ref(app_name_iter, -1);
            }
            ao2_iterator_destroy(&iter);
        }
    }
 
    return 0;
}
 
/*!
 * \brief Get the winner app name for a broadcast channel
 * \param channel_id The unique ID of the channel
 * \return A copy of the winner app name (caller must free with ast_free),
 *         or NULL if not claimed or not found
 */
char *AST_OPTIONAL_API_NAME(stasis_app_broadcast_winner)(const char *channel_id)
{
    RAII_VAR(struct stasis_broadcast_ctx *, ctx, NULL, ao2_cleanup);
    char *winner = NULL;
 
    if (ast_strlen_zero(channel_id)) {
        return NULL;
    }
 
    ctx = ao2_find(broadcast_contexts, channel_id, OBJ_SEARCH_KEY);
    if (!ctx) {
        return NULL;
    }
 
    ao2_lock(ctx);
    if (ctx->claimed) {
        winner = ast_strdup(ctx->winner_app);
    }
    /* Mark the broadcast as finished so no new claims can succeed.
     * This closes the race window between reading the winner and
     * the subsequent broadcast_cleanup call. */
    ctx->finished = 1;
    ao2_unlock(ctx);
 
    return winner;
}
 
/*!
 * \brief Wait for a broadcast channel to be claimed
 *
 * Blocks until the channel is claimed, the timeout expires, or the
 * channel hangs up.  The hangup check runs every
 * #BROADCAST_POLL_INTERVAL_MS so that a dead channel does not tie up
 * a PBX thread for the full timeout period.
 *
 * \param chan The channel
 * \param timeout_ms Maximum time to wait in milliseconds
 * \return 0 if claimed within timeout, -1 otherwise
 */
int AST_OPTIONAL_API_NAME(stasis_app_broadcast_wait)(struct ast_channel *chan, int timeout_ms)
{
    RAII_VAR(struct stasis_broadcast_ctx *, ctx, NULL, ao2_cleanup);
    const char *channel_id;
    struct timeval deadline;
    int result = -1;
 
    if (!chan) {
        return -1;
    }
 
    channel_id = ast_channel_uniqueid(chan);
 
    if (!broadcast_contexts) {
        return -1;
    }
 
    ctx = ao2_find(broadcast_contexts, channel_id, OBJ_SEARCH_KEY);
    if (!ctx) {
        ast_log(LOG_WARNING, "No broadcast context for channel %s\n", channel_id);
        return -1;
    }
 
    /* Cap excessive timeouts to prevent arithmetic overflow */
    if (timeout_ms < 0) {
        timeout_ms = 0;
    } else if (timeout_ms > MAX_BROADCAST_TIMEOUT_MS) {
        timeout_ms = MAX_BROADCAST_TIMEOUT_MS;
    }
 
    /* Calculate absolute deadline */
    deadline = ast_tvadd(ast_tvnow(),
        ast_tv(timeout_ms / 1000, (timeout_ms % 1000) * 1000));
 
    ao2_lock(ctx);
    while (!ctx->claimed) {
        struct timeval now;
        struct timespec poll_spec;
        long remaining_ms;
        long poll_ms;
        int wait_result;
 
        /* Check for hangup so we don't block on a dead channel */
        if (ast_check_hangup(chan)) {
            ast_debug(3, "Channel %s hung up during broadcast wait\n",
                channel_id);
            break;
        }
 
        /* Check if we've passed the overall deadline */
        now = ast_tvnow();
        remaining_ms = ast_tvdiff_ms(deadline, now);
        if (remaining_ms <= 0) {
            ast_debug(3, "Broadcast timeout for channel %s after %dms\n",
                channel_id, timeout_ms);
            break;
        }
 
        /* Sleep for the shorter of the remaining time and the poll interval */
        poll_ms = remaining_ms;
        if (poll_ms > BROADCAST_POLL_INTERVAL_MS) {
            poll_ms = BROADCAST_POLL_INTERVAL_MS;
        }
 
        poll_spec.tv_sec = now.tv_sec + (poll_ms / 1000);
        poll_spec.tv_nsec = (long)(now.tv_usec) * 1000L
            + (long)(poll_ms % 1000) * 1000000L;
        while (poll_spec.tv_nsec >= 1000000000) {
            poll_spec.tv_sec++;
            poll_spec.tv_nsec -= 1000000000;
        }
 
        wait_result = ast_cond_timedwait(&ctx->cond, ao2_object_get_lockaddr(ctx), &poll_spec);
        if (wait_result != 0 && wait_result != ETIMEDOUT) {
            ast_log(LOG_WARNING,
                "Channel %s: unexpected error waiting for claim: %s (%d)\n",
                channel_id, strerror(wait_result), wait_result);
            break;
        }
        /* Loop back: re-check claimed, then hangup, then deadline */
    }
 
    if (ctx->claimed) {
        ast_debug(1, "Channel %s claimed by %s\n",
            channel_id, ctx->winner_app);
        result = 0;
    }
    ao2_unlock(ctx);
 
    return result;
}
 
/*!
 * \brief Clean up broadcast context for a channel
 *
 * This is the normal-path cleanup called by the dialplan application
 * after the broadcast completes.  The channel datastore destructor
 * (broadcast_datastore_destroy) also unlinks the context as a safety
 * net for abnormal teardown; ao2_unlink is idempotent so the double
 * call is harmless.
 *
 * \param channel_id The unique ID of the channel
 */
void AST_OPTIONAL_API_NAME(stasis_app_broadcast_cleanup)(const char *channel_id)
{
    RAII_VAR(struct stasis_broadcast_ctx *, ctx, NULL, ao2_cleanup);
 
    if (ast_strlen_zero(channel_id) || !broadcast_contexts) {
        return;
    }
 
    ctx = ao2_find(broadcast_contexts, channel_id, OBJ_SEARCH_KEY | OBJ_UNLINK);
    if (ctx) {
        ast_debug(3, "Cleaning up broadcast context for %s\n", channel_id);
    }
}
 
static int load_module(void)
{
    broadcast_contexts = ao2_container_alloc_hash(AO2_ALLOC_OPT_LOCK_MUTEX, 0,
        BROADCAST_BUCKETS, stasis_broadcast_ctx_hash_fn, NULL, stasis_broadcast_ctx_cmp_fn);
 
    if (!broadcast_contexts) {
        return AST_MODULE_LOAD_DECLINE;
    }
 
    ast_debug(1, "Stasis broadcast module loaded\n");
    return AST_MODULE_LOAD_SUCCESS;
}
 
static int unload_module(void)
{
    /* NULL the global pointer before releasing the reference so that
     * concurrent lookups see NULL (safe) rather than a freed pointer. */
    {
        struct ao2_container *old_contexts = broadcast_contexts;
        broadcast_contexts = NULL;
        ao2_cleanup(old_contexts);
    }
 
    ast_debug(1, "Stasis broadcast module unloaded\n");
    return 0;
}
 
AST_MODULE_INFO(ASTERISK_GPL_KEY, AST_MODFLAG_GLOBAL_SYMBOLS | AST_MODFLAG_LOAD_ORDER,
    "Stasis application broadcast",
    .support_level = AST_MODULE_SUPPORT_EXTENDED,
    .load = load_module,
    .unload = unload_module,
    .requires = "res_stasis,res_ari,http",
    .load_pri = AST_MODPRI_APP_DEPEND - 1,
);