v0.0.55 - Defer trigger loading until self-skill EOSE and add trigger subscription debug instrumentation
This commit is contained in:
Your Name
+128
-60
@@ -12,6 +12,11 @@
|
||||
#include "debug.h"
|
||||
#include "nostr_handler.h"
|
||||
|
||||
typedef struct {
|
||||
trigger_manager_t* mgr;
|
||||
char skill_slug[TRIGGER_SKILL_SLUG_MAX];
|
||||
} trigger_subscription_ctx_t;
|
||||
|
||||
static int clamp_enabled(int enabled) {
|
||||
return enabled ? 1 : 0;
|
||||
}
|
||||
@@ -220,6 +225,107 @@ static int maybe_fire_trigger_locked(trigger_manager_t* mgr, int index, cJSON* e
|
||||
return 1;
|
||||
}
|
||||
|
||||
static void on_trigger_subscription_eose(cJSON** events, int event_count, void* user_data) {
|
||||
(void)events;
|
||||
(void)event_count;
|
||||
(void)user_data;
|
||||
}
|
||||
|
||||
static void close_trigger_subscription_locked(active_trigger_t* t) {
|
||||
if (!t) {
|
||||
return;
|
||||
}
|
||||
|
||||
if (t->subscription) {
|
||||
(void)nostr_handler_close_subscription(t->subscription);
|
||||
t->subscription = NULL;
|
||||
}
|
||||
|
||||
if (t->subscription_ctx) {
|
||||
free(t->subscription_ctx);
|
||||
t->subscription_ctx = NULL;
|
||||
}
|
||||
}
|
||||
|
||||
static void on_trigger_subscription_event(cJSON* event, const char* relay_url, void* user_data) {
|
||||
trigger_subscription_ctx_t* ctx = (trigger_subscription_ctx_t*)user_data;
|
||||
if (!ctx || !ctx->mgr || !event) {
|
||||
return;
|
||||
}
|
||||
|
||||
trigger_manager_t* mgr = ctx->mgr;
|
||||
pthread_mutex_lock(&mgr->mutex);
|
||||
|
||||
int idx = find_trigger_index_locked(mgr, ctx->skill_slug);
|
||||
if (idx >= 0) {
|
||||
(void)maybe_fire_trigger_locked(mgr, idx, event, relay_url);
|
||||
}
|
||||
|
||||
pthread_mutex_unlock(&mgr->mutex);
|
||||
}
|
||||
|
||||
static int register_trigger_subscription_locked(trigger_manager_t* mgr, active_trigger_t* t) {
|
||||
if (!mgr || !t) {
|
||||
return -1;
|
||||
}
|
||||
|
||||
close_trigger_subscription_locked(t);
|
||||
|
||||
if (!t->enabled || t->filter_json[0] == '\0') {
|
||||
return 0;
|
||||
}
|
||||
|
||||
cJSON* filter = cJSON_Parse(t->filter_json);
|
||||
if (!filter || !cJSON_IsObject(filter)) {
|
||||
cJSON_Delete(filter);
|
||||
return -1;
|
||||
}
|
||||
|
||||
cJSON* since = cJSON_GetObjectItemCaseSensitive(filter, "since");
|
||||
if (!since || !cJSON_IsNumber(since)) {
|
||||
time_t now = time(NULL);
|
||||
time_t default_since = now > 30 ? now - 30 : 0;
|
||||
cJSON_AddNumberToObject(filter, "since", (double)default_since);
|
||||
}
|
||||
|
||||
cJSON* limit = cJSON_GetObjectItemCaseSensitive(filter, "limit");
|
||||
if (!limit || !cJSON_IsNumber(limit)) {
|
||||
cJSON_AddNumberToObject(filter, "limit", 200);
|
||||
}
|
||||
|
||||
trigger_subscription_ctx_t* ctx = (trigger_subscription_ctx_t*)calloc(1, sizeof(*ctx));
|
||||
if (!ctx) {
|
||||
cJSON_Delete(filter);
|
||||
return -1;
|
||||
}
|
||||
|
||||
ctx->mgr = mgr;
|
||||
snprintf(ctx->skill_slug, sizeof(ctx->skill_slug), "%s", t->skill_slug);
|
||||
|
||||
nostr_pool_subscription_t* sub = nostr_handler_subscribe_with_filter(
|
||||
filter,
|
||||
on_trigger_subscription_event,
|
||||
on_trigger_subscription_eose,
|
||||
ctx,
|
||||
0,
|
||||
1,
|
||||
NOSTR_POOL_EOSE_FULL_SET,
|
||||
30,
|
||||
120);
|
||||
|
||||
cJSON_Delete(filter);
|
||||
|
||||
if (!sub) {
|
||||
free(ctx);
|
||||
return -1;
|
||||
}
|
||||
|
||||
t->subscription = sub;
|
||||
t->subscription_ctx = ctx;
|
||||
|
||||
return 0;
|
||||
}
|
||||
|
||||
int trigger_manager_init(trigger_manager_t* mgr, didactyl_config_t* cfg) {
|
||||
if (!mgr || !cfg) {
|
||||
return -1;
|
||||
@@ -433,6 +539,15 @@ int trigger_manager_add(trigger_manager_t* mgr,
|
||||
t->enabled = clamp_enabled(enabled);
|
||||
t->last_fired = 0;
|
||||
t->last_seen_created_at = 0;
|
||||
t->subscription = NULL;
|
||||
t->subscription_ctx = NULL;
|
||||
|
||||
if (register_trigger_subscription_locked(mgr, t) != 0) {
|
||||
pthread_mutex_unlock(&mgr->mutex);
|
||||
DEBUG_WARN("[didactyl] trigger add rejected: failed to create subscription slug=%s", skill_slug);
|
||||
memset(t, 0, sizeof(*t));
|
||||
return -1;
|
||||
}
|
||||
|
||||
mgr->count++;
|
||||
|
||||
@@ -455,6 +570,8 @@ int trigger_manager_remove(trigger_manager_t* mgr, const char* skill_slug) {
|
||||
return 0;
|
||||
}
|
||||
|
||||
close_trigger_subscription_locked(&mgr->triggers[idx]);
|
||||
|
||||
if (idx < mgr->count - 1) {
|
||||
memmove(&mgr->triggers[idx],
|
||||
&mgr->triggers[idx + 1],
|
||||
@@ -498,6 +615,12 @@ int trigger_manager_update(trigger_manager_t* mgr,
|
||||
t->action_type = action_type;
|
||||
t->enabled = clamp_enabled(enabled);
|
||||
|
||||
if (register_trigger_subscription_locked(mgr, t) != 0) {
|
||||
pthread_mutex_unlock(&mgr->mutex);
|
||||
DEBUG_WARN("[didactyl] trigger update failed to (re)subscribe slug=%s", skill_slug);
|
||||
return -1;
|
||||
}
|
||||
|
||||
pthread_mutex_unlock(&mgr->mutex);
|
||||
|
||||
DEBUG_INFO("[didactyl] trigger updated slug=%s action=%d enabled=%d", skill_slug, (int)action_type, clamp_enabled(enabled));
|
||||
@@ -522,66 +645,7 @@ int trigger_manager_active_count(trigger_manager_t* mgr) {
|
||||
}
|
||||
|
||||
int trigger_manager_poll(trigger_manager_t* mgr) {
|
||||
if (!mgr || !mgr->cfg || !mgr->cfg->triggers.enabled) {
|
||||
return 0;
|
||||
}
|
||||
|
||||
time_t now = time(NULL);
|
||||
if (mgr->last_poll_at > 0 && (now - mgr->last_poll_at) < 10) {
|
||||
return 0;
|
||||
}
|
||||
mgr->last_poll_at = now;
|
||||
|
||||
pthread_mutex_lock(&mgr->mutex);
|
||||
int count = mgr->count;
|
||||
|
||||
for (int i = 0; i < count; i++) {
|
||||
active_trigger_t snapshot = mgr->triggers[i];
|
||||
if (!snapshot.enabled || snapshot.filter_json[0] == '\0') {
|
||||
continue;
|
||||
}
|
||||
|
||||
cJSON* filter = cJSON_Parse(snapshot.filter_json);
|
||||
if (!filter || !cJSON_IsObject(filter)) {
|
||||
cJSON_Delete(filter);
|
||||
continue;
|
||||
}
|
||||
|
||||
time_t since = snapshot.last_seen_created_at > 0 ? snapshot.last_seen_created_at + 1 : now - 30;
|
||||
if (since < 0) since = 0;
|
||||
cJSON_AddNumberToObject(filter, "since", (double)since);
|
||||
cJSON_AddNumberToObject(filter, "limit", 8);
|
||||
|
||||
char* events_json = nostr_handler_query_json(filter, 1200);
|
||||
cJSON_Delete(filter);
|
||||
if (!events_json) {
|
||||
continue;
|
||||
}
|
||||
|
||||
cJSON* events = cJSON_Parse(events_json);
|
||||
free(events_json);
|
||||
if (!events || !cJSON_IsArray(events)) {
|
||||
cJSON_Delete(events);
|
||||
continue;
|
||||
}
|
||||
|
||||
int n = cJSON_GetArraySize(events);
|
||||
for (int e = 0; e < n; e++) {
|
||||
cJSON* ev = cJSON_GetArrayItem(events, e);
|
||||
if (!ev || !cJSON_IsObject(ev)) continue;
|
||||
|
||||
int idx = find_trigger_index_locked(mgr, snapshot.skill_slug);
|
||||
if (idx < 0) {
|
||||
continue;
|
||||
}
|
||||
|
||||
(void)maybe_fire_trigger_locked(mgr, idx, ev, NULL);
|
||||
}
|
||||
|
||||
cJSON_Delete(events);
|
||||
}
|
||||
|
||||
pthread_mutex_unlock(&mgr->mutex);
|
||||
(void)mgr;
|
||||
return 0;
|
||||
}
|
||||
|
||||
@@ -627,6 +691,7 @@ char* trigger_manager_status_json(trigger_manager_t* mgr) {
|
||||
cJSON_AddStringToObject(item, "filter_json", t->filter_json);
|
||||
cJSON_AddStringToObject(item, "action", t->action_type == TRIGGER_ACTION_TEMPLATE ? "template" : "llm");
|
||||
cJSON_AddBoolToObject(item, "enabled", t->enabled ? 1 : 0);
|
||||
cJSON_AddBoolToObject(item, "subscribed", t->subscription ? 1 : 0);
|
||||
cJSON_AddNumberToObject(item, "last_fired", (double)t->last_fired);
|
||||
cJSON_AddNumberToObject(item, "last_seen_created_at", (double)t->last_seen_created_at);
|
||||
cJSON_AddItemToArray(arr, item);
|
||||
@@ -648,6 +713,9 @@ void trigger_manager_cleanup(trigger_manager_t* mgr) {
|
||||
}
|
||||
|
||||
pthread_mutex_lock(&mgr->mutex);
|
||||
for (int i = 0; i < mgr->count; i++) {
|
||||
close_trigger_subscription_locked(&mgr->triggers[i]);
|
||||
}
|
||||
free(mgr->triggers);
|
||||
mgr->triggers = NULL;
|
||||
mgr->count = 0;
|
||||
|
||||
Reference in New Issue
Block a user