commit 928f22f

Email notifications have been moved to the global queue.

 1@@ -839,7 +839,7 @@ void notify(snac *snac, char *type, char *utype, char *actor, char *msg)
 2         body = xs_str_cat(body, s1);
 3     }
 4 
 5-    enqueue_email(snac, body, 0);
 6+    enqueue_email(body, 0);
 7 }
 8 
 9 
10@@ -1137,26 +1137,6 @@ void process_user_queue_item(snac *snac, xs_dict *q_item)
11             }
12         }
13     }
14-    else
15-    if (strcmp(type, "email") == 0) {
16-        /* send this email */
17-        xs_str *msg = xs_dict_get(q_item, "message");
18-        int retries = xs_number_get(xs_dict_get(q_item, "retries"));
19-
20-        if (!send_email(msg))
21-            snac_debug(snac, 1, xs_fmt("email message sent"));
22-        else {
23-            if (retries > queue_retry_max)
24-                snac_log(snac, xs_fmt("process_queue email giving up (errno: %d)", errno));
25-            else {
26-                /* requeue */
27-                snac_log(snac, xs_fmt(
28-                    "process_queue email requeue #%d (errno: %d)", retries + 1, errno));
29-
30-                enqueue_email(snac, msg, retries + 1);
31-            }
32-        }
33-    }
34 }
35 
36 
37@@ -1184,6 +1164,30 @@ void process_user_queue(snac *snac)
38 void process_queue_item(xs_dict *q_item)
39 /* processes an item from the global queue */
40 {
41+    char *type = xs_dict_get(q_item, "type");
42+    int queue_retry_max = xs_number_get(xs_dict_get(srv_config, "queue_retry_max"));
43+
44+    if (strcmp(type, "email") == 0) {
45+        /* send this email */
46+        xs_str *msg = xs_dict_get(q_item, "message");
47+        int retries = xs_number_get(xs_dict_get(q_item, "retries"));
48+
49+        if (!send_email(msg))
50+            srv_debug(1, xs_fmt("email message sent"));
51+        else {
52+            retries++;
53+
54+            if (retries > queue_retry_max)
55+                srv_log(xs_fmt("process_queue email giving up (errno: %d)", errno));
56+            else {
57+                /* requeue */
58+                srv_log(xs_fmt(
59+                    "process_queue email requeue #%d (errno: %d)", retries, errno));
60+
61+                enqueue_email(msg, retries);
62+            }
63+        }
64+    }
65 }
66 
67 

 1@@ -1389,16 +1389,16 @@ void enqueue_output_by_actor(snac *snac, xs_dict *msg, xs_str *actor, int retrie
 2 }
 3 
 4 
 5-void enqueue_email(snac *snac, xs_str *msg, int retries)
 6+void enqueue_email(xs_str *msg, int retries)
 7 /* enqueues an email message to be sent */
 8 {
 9     xs *qmsg   = _new_qmsg("email", msg, retries);
10     char *ntid = xs_dict_get(qmsg, "ntid");
11-    xs *fn     = xs_fmt("%s/queue/%s.json", snac->basedir, ntid);
12+    xs *fn     = xs_fmt("%s/queue/%s.json", srv_basedir, ntid);
13 
14     qmsg = _enqueue_put(fn, qmsg);
15 
16-    snac_debug(snac, 1, xs_fmt("enqueue_email %d", retries));
17+    srv_debug(1, xs_fmt("enqueue_email %d", retries));
18 }
19 
20 

 1@@ -280,6 +280,9 @@ static void *queue_thread(void *arg)
 2             }
 3         }
 4 
 5+        /* global queue */
 6+        process_queue();
 7+
 8         /* time to purge? */
 9         if ((t = time(NULL)) > purge_time) {
10             pthread_t pth;

 1@@ -126,10 +126,11 @@ d_char *history_list(snac *snac);
 2 void enqueue_input(snac *snac, xs_dict *msg, xs_dict *req, int retries);
 3 void enqueue_output(snac *snac, xs_dict *msg, xs_str *inbox, int retries);
 4 void enqueue_output_by_actor(snac *snac, xs_dict *msg, xs_str *actor, int retries);
 5-void enqueue_email(snac *snac, xs_str *msg, int retries);
 6+void enqueue_email(xs_str *msg, int retries);
 7 void enqueue_message(snac *snac, char *msg);
 8 
 9 xs_list *user_queue(snac *snac);
10+xs_list *queue(void);
11 xs_dict *dequeue(const char *fn);
12 
13 void purge(snac *snac);
14@@ -165,7 +166,11 @@ int send_to_inbox(snac *snac, char *inbox, char *msg, d_char **payload, int *p_s
15 d_char *get_actor_inbox(snac *snac, char *actor);
16 int send_to_actor(snac *snac, char *actor, char *msg, d_char **payload, int *p_size, int timeout);
17 int is_msg_public(snac *snac, char *msg);
18+
19 void process_user_queue(snac *snac);
20+
21+void process_queue(void);
22+
23 void post(snac *snac, char *msg);
24 int activitypub_get_handler(d_char *req, char *q_path,
25                             char **body, int *b_size, char **ctype);