1 files changed, 47 insertions, 26 deletions

diff --git a/activitypub.c b/activitypub.c
index 55a245e..d7f5b37 100644
--- a/activitypub.c
+++ b/activitypub.c
@@ -2052,45 +2052,66 @@ void process_queue_item(xs_dict *q_item)
     }
     else
     if (strcmp(type, "input") == 0) {
-        /* redistribute the input message to all users */
-        char *ntid   = xs_dict_get(q_item, "ntid");
-        xs *tmpfn    = xs_fmt("%s/tmp/%s.json", srv_basedir, ntid);
         xs_dict *msg = xs_dict_get(q_item, "message");
-        FILE *f;
+        xs_dict *req = xs_dict_get(q_item, "req");
+        int retries  = xs_number_get(xs_dict_get(q_item, "retries"));
+
+        /* do some instance-level checks */
+        int r = process_input_message(NULL, msg, req);
 
-        if ((f = fopen(tmpfn, "w")) != NULL) {
-            xs_json_dump(q_item, 4, f);
-            fclose(f);
+        if (r == 0) {
+            /* transient error? retry */
+            int queue_retry_max = xs_number_get(xs_dict_get(srv_config, "queue_retry_max"));
+
+            if (retries > queue_retry_max)
+                srv_log(xs_fmt("shared input giving up"));
+            else {
+                /* reenqueue */
+                enqueue_shared_input(msg, req, retries + 1);
+                srv_log(xs_fmt("shared input requeue #%d", retries + 1));
+            }
         }
+        else
+        if (r == 2) {
+            /* redistribute the input message to all users */
+            char *ntid = xs_dict_get(q_item, "ntid");
+            xs *tmpfn  = xs_fmt("%s/tmp/%s.json", srv_basedir, ntid);
+            FILE *f;
+
+            if ((f = fopen(tmpfn, "w")) != NULL) {
+                xs_json_dump(q_item, 4, f);
+                fclose(f);
+            }
 
-        xs *users = user_list();
-        xs_list *p = users;
-        char *v;
-        int cnt = 0;
+            xs *users = user_list();
+            xs_list *p = users;
+            char *v;
+            int cnt = 0;
 
-        while (xs_list_iter(&p, &v)) {
-            snac user;
+            while (xs_list_iter(&p, &v)) {
+                snac user;
 
-            if (user_open(&user, v)) {
-                if (is_msg_for_me(&user, msg)) {
-                    xs *fn = xs_fmt("%s/queue/%s.json", user.basedir, ntid);
+                if (user_open(&user, v)) {
+                    if (is_msg_for_me(&user, msg)) {
+                        xs *fn = xs_fmt("%s/queue/%s.json", user.basedir, ntid);
 
-                    snac_debug(&user, 1, xs_fmt("enqueue_input (from shared inbox) %s", fn));
+                        snac_debug(&user, 1, xs_fmt("enqueue_input (from shared inbox) %s", fn));
 
-                    if (link(tmpfn, fn) < 0)
-                        srv_log(xs_fmt("link(%s, %s) error", tmpfn, fn));
+                        if (link(tmpfn, fn) < 0)
+                            srv_log(xs_fmt("link(%s, %s) error", tmpfn, fn));
 
-                    cnt++;
-                }
+                        cnt++;
+                    }
 
-                user_free(&user);
+                    user_free(&user);
+                }
             }
-        }
 
-        unlink(tmpfn);
+            unlink(tmpfn);
 
-        if (cnt == 0)
-            srv_debug(1, xs_fmt("no valid recipients for %s", tmpfn));
+            if (cnt == 0)
+                srv_debug(1, xs_fmt("no valid recipients for %s", tmpfn));
+        }
     }
     else
         srv_log(xs_fmt("unexpected q_item type '%s'", type));