snac2/data.c

901 lines
20 KiB
C
Raw Normal View History

2022-09-19 23:41:30 +03:00
/* snac - A simple, minimalistic ActivityPub instance */
/* copyright (c) 2022 grunfink - MIT license */
#include "xs.h"
#include "xs_io.h"
#include "xs_json.h"
2022-09-20 10:39:28 +03:00
#include "xs_openssl.h"
2022-09-19 23:41:30 +03:00
#include "snac.h"
2022-09-20 13:56:21 +03:00
#include <time.h>
2022-09-20 00:33:11 +03:00
#include <glob.h>
2022-09-20 13:56:21 +03:00
#include <sys/stat.h>
2022-09-20 00:33:11 +03:00
2022-09-19 23:41:30 +03:00
int srv_open(char *basedir)
/* opens a server */
{
int ret = 0;
xs *cfg_file = NULL;
FILE *f;
2022-09-20 10:39:28 +03:00
d_char *error = NULL;
2022-09-19 23:41:30 +03:00
srv_basedir = xs_str_new(basedir);
2022-09-20 10:39:28 +03:00
if (xs_endswith(srv_basedir, "/"))
srv_basedir = xs_crop(srv_basedir, 0, -1);
2022-09-19 23:41:30 +03:00
cfg_file = xs_fmt("%s/server.json", basedir);
if ((f = fopen(cfg_file, "r")) == NULL)
2022-09-20 10:39:28 +03:00
error = xs_fmt("error opening '%s'", cfg_file);
2022-09-19 23:41:30 +03:00
else {
xs *cfg_data;
/* read full config file */
cfg_data = xs_readall(f);
/* parse */
srv_config = xs_json_loads(cfg_data);
if (srv_config == NULL)
2022-09-20 10:39:28 +03:00
error = xs_fmt("cannot parse '%s'", cfg_file);
2022-09-19 23:41:30 +03:00
else {
char *host;
char *prefix;
char *dbglvl;
host = xs_dict_get(srv_config, "host");
prefix = xs_dict_get(srv_config, "prefix");
dbglvl = xs_dict_get(srv_config, "dbglevel");
if (host == NULL || prefix == NULL)
2022-09-20 10:39:28 +03:00
error = xs_str_new("cannot get server data");
2022-09-19 23:41:30 +03:00
else {
srv_baseurl = xs_fmt("https://%s%s", host, prefix);
dbglevel = (int) xs_number_get(dbglvl);
if ((dbglvl = getenv("DEBUG")) != NULL) {
dbglevel = atoi(dbglvl);
2022-09-20 10:39:28 +03:00
error = xs_fmt("DEBUG level set to %d from environment", dbglevel);
2022-09-19 23:41:30 +03:00
}
ret = 1;
}
}
}
2022-09-20 10:39:28 +03:00
if (ret == 0 && error != NULL)
srv_log(error);
2022-09-19 23:41:30 +03:00
return ret;
}
2022-09-20 00:33:11 +03:00
void user_free(snac *snac)
2022-09-19 23:41:30 +03:00
/* frees a user snac */
{
free(snac->uid);
free(snac->basedir);
free(snac->config);
free(snac->key);
free(snac->actor);
}
2022-09-20 00:03:18 +03:00
int user_open(snac *snac, char *uid)
2022-09-19 23:41:30 +03:00
/* opens a user */
{
int ret = 0;
memset(snac, '\0', sizeof(struct _snac));
if (validate_uid(uid)) {
xs *cfg_file;
FILE *f;
snac->uid = xs_str_new(uid);
snac->basedir = xs_fmt("%s/user/%s", srv_basedir, uid);
cfg_file = xs_fmt("%s/user.json", snac->basedir);
if ((f = fopen(cfg_file, "r")) != NULL) {
xs *cfg_data;
/* read full config file */
cfg_data = xs_readall(f);
fclose(f);
if ((snac->config = xs_json_loads(cfg_data)) != NULL) {
xs *key_file = xs_fmt("%s/key.json", snac->basedir);
if ((f = fopen(key_file, "r")) != NULL) {
xs *key_data;
key_data = xs_readall(f);
fclose(f);
if ((snac->key = xs_json_loads(key_data)) != NULL) {
snac->actor = xs_fmt("%s/%s", srv_baseurl, uid);
ret = 1;
}
else
srv_log(xs_fmt("cannot parse '%s'", key_file));
}
else
srv_log(xs_fmt("error opening '%s'", key_file));
}
else
srv_log(xs_fmt("cannot parse '%s'", cfg_file));
}
else
2022-09-25 10:55:33 +03:00
srv_debug(2, xs_fmt("error opening '%s'", cfg_file));
2022-09-19 23:41:30 +03:00
}
else
srv_log(xs_fmt("invalid user '%s'", uid));
if (!ret)
2022-09-20 00:33:11 +03:00
user_free(snac);
2022-09-19 23:41:30 +03:00
return ret;
}
2022-09-20 00:33:11 +03:00
d_char *user_list(void)
/* returns the list of user ids */
{
d_char *list;
xs *spec;
glob_t globbuf;
globbuf.gl_offs = 1;
list = xs_list_new();
2022-09-20 10:39:28 +03:00
spec = xs_fmt("%s/user/" "*", srv_basedir);
2022-09-20 00:33:11 +03:00
if (glob(spec, 0, NULL, &globbuf) == 0) {
int n;
char *p;
for (n = 0; (p = globbuf.gl_pathv[n]) != NULL; n++) {
if ((p = strrchr(p, '/')) != NULL)
list = xs_list_append(list, p + 1);
}
}
globfree(&globbuf);
return list;
}
2022-09-20 10:39:28 +03:00
float mtime(char *fn)
/* returns the mtime of a file or directory, or 0.0 */
{
struct stat st;
float r = 0.0;
if (stat(fn, &st) != -1)
r = (float)st.st_mtim.tv_sec;
return r;
}
d_char *_follower_fn(snac *snac, char *actor)
{
xs *md5 = xs_md5_hex(actor, strlen(actor));
return xs_fmt("%s/followers/%s.json", snac->basedir, md5);
}
int follower_add(snac *snac, char *actor, char *msg)
/* adds a follower */
{
int ret = 201; /* created */
xs *fn = _follower_fn(snac, actor);
FILE *f;
if ((f = fopen(fn, "w")) != NULL) {
xs *j = xs_json_dumps_pp(msg, 4);
fwrite(j, 1, strlen(j), f);
fclose(f);
}
else
ret = 500;
snac_debug(snac, 2, xs_fmt("follower_add %s %s", actor, fn));
return ret;
}
int follower_del(snac *snac, char *actor)
/* deletes a follower */
{
xs *fn = _follower_fn(snac, actor);
unlink(fn);
snac_debug(snac, 2, xs_fmt("follower_del %s %s", actor, fn));
return 200;
}
int follower_check(snac *snac, char *actor)
/* checks if someone is a follower */
{
xs *fn = _follower_fn(snac, actor);
return !!(mtime(fn) != 0.0);
}
d_char *follower_list(snac *snac)
/* returns the list of followers */
{
d_char *list;
xs *spec;
glob_t globbuf;
list = xs_list_new();
spec = xs_fmt("%s/followers/" "*.json", snac->basedir);
if (glob(spec, 0, NULL, &globbuf) == 0) {
int n;
2022-09-20 11:49:24 +03:00
char *fn;
2022-09-20 10:39:28 +03:00
2022-09-20 11:49:24 +03:00
for (n = 0; (fn = globbuf.gl_pathv[n]) != NULL; n++) {
2022-09-20 10:39:28 +03:00
FILE *f;
2022-09-20 11:49:24 +03:00
if ((f = fopen(fn, "r")) != NULL) {
2022-09-20 10:48:13 +03:00
xs *j = xs_readall(f);
xs *o = xs_json_loads(j);
if (o != NULL)
list = xs_list_append(list, o);
fclose(f);
2022-09-20 10:39:28 +03:00
}
}
}
globfree(&globbuf);
return list;
}
2022-09-20 11:02:00 +03:00
2022-09-20 11:49:24 +03:00
d_char *_timeline_find_fn(snac *snac, char *id)
2022-09-20 11:02:00 +03:00
/* returns the file name of a timeline entry by its id */
{
xs *md5 = xs_md5_hex(id, strlen(id));
xs *spec = xs_fmt("%s/timeline/" "*-%s.json", snac->basedir, md5);
glob_t globbuf;
d_char *fn = NULL;
2022-09-20 13:56:21 +03:00
if (glob(spec, 0, NULL, &globbuf) == 0 && globbuf.gl_pathc) {
2022-09-20 11:02:00 +03:00
/* get just the first file */
fn = xs_str_new(globbuf.gl_pathv[0]);
}
globfree(&globbuf);
return fn;
}
2022-09-25 19:28:15 +03:00
int timeline_here(snac *snac, char *id)
/* checks if an object is already downloaded */
{
xs *fn = _timeline_find_fn(snac, id);
return fn != NULL;
}
2022-09-20 11:49:24 +03:00
d_char *timeline_find(snac *snac, char *id)
/* gets a message from the timeline by id */
2022-09-20 11:02:00 +03:00
{
2022-09-20 11:49:24 +03:00
xs *fn = _timeline_find_fn(snac, id);
2022-09-20 11:02:00 +03:00
xs *msg = NULL;
if (fn != NULL) {
FILE *f;
if ((f = fopen(fn, "r")) != NULL) {
xs *j = xs_readall(f);
msg = xs_json_loads(j);
fclose(f);
}
}
return msg;
}
void timeline_del(snac *snac, char *id)
/* deletes a message from the timeline */
{
2022-09-20 11:49:24 +03:00
xs *fn = _timeline_find_fn(snac, id);
2022-09-20 11:02:00 +03:00
if (fn != NULL) {
xs *lfn = NULL;
unlink(fn);
snac_debug(snac, 1, xs_fmt("timeline_del %s", id));
/* try to delete also from the local timeline */
lfn = xs_replace(fn, "/timeline/", "/local/");
if (unlink(lfn) != -1)
snac_debug(snac, 1, xs_fmt("timeline_del (local) %s", id));
}
}
2022-09-20 11:49:24 +03:00
d_char *timeline_get(snac *snac, char *fn)
/* gets a timeline entry by file name */
{
d_char *d = NULL;
FILE *f;
if ((f = fopen(fn, "r")) != NULL) {
xs *j = xs_readall(f);
d = xs_json_loads(j);
fclose(f);
}
return d;
}
d_char *timeline_list(snac *snac)
/* returns a list of the timeline filenames */
{
d_char *list;
xs *spec = xs_fmt("%s/timeline/" "*.json", snac->basedir);
glob_t globbuf;
int max;
/* maximum number of items in the timeline */
max = xs_number_get(xs_dict_get(srv_config, "max_timeline_entries"));
list = xs_list_new();
/* get the list in reverse order */
if (glob(spec, 0, NULL, &globbuf) == 0) {
int n;
2022-09-20 13:56:21 +03:00
if (max > globbuf.gl_pathc)
max = globbuf.gl_pathc;
2022-09-20 11:49:24 +03:00
for (n = 0; n < max; n++) {
2022-09-20 13:56:21 +03:00
char *fn = globbuf.gl_pathv[globbuf.gl_pathc - n - 1];
2022-09-20 11:49:24 +03:00
list = xs_list_append(list, fn);
}
}
globfree(&globbuf);
return list;
}
2022-09-20 12:16:24 +03:00
2022-09-23 10:50:38 +03:00
d_char *_timeline_new_fn(snac *snac, char *id)
/* creates a new filename */
2022-09-22 15:46:23 +03:00
{
2022-09-23 10:50:38 +03:00
xs *ntid = tid(0);
xs *md5 = xs_md5_hex(id, strlen(id));
2022-09-22 15:46:23 +03:00
2022-09-23 10:50:38 +03:00
return xs_fmt("%s/timeline/%s-%s.json", snac->basedir, ntid, md5);
2022-09-22 15:46:23 +03:00
}
void _timeline_write(snac *snac, char *id, char *msg, char *parent, char *referrer)
2022-09-25 18:42:39 +03:00
/* writes a timeline entry and refreshes the ancestors */
2022-09-20 12:16:24 +03:00
{
2022-09-23 10:50:38 +03:00
xs *fn = _timeline_new_fn(snac, id);
2022-09-25 18:42:39 +03:00
FILE *f;
2022-09-20 12:16:24 +03:00
if ((f = fopen(fn, "w")) != NULL) {
xs *j = xs_json_dumps_pp(msg, 4);
fwrite(j, strlen(j), 1, f);
fclose(f);
2022-09-25 18:42:39 +03:00
snac_debug(snac, 1, xs_fmt("_timeline_write %s %s", id, fn));
2022-09-20 12:16:24 +03:00
}
2022-09-23 10:50:38 +03:00
/* related to this user? link to local timeline */
if (xs_startswith(id, snac->actor) ||
(!xs_is_null(parent) && xs_startswith(parent, snac->actor)) ||
(!xs_is_null(referrer) && xs_startswith(referrer, snac->actor))) {
2022-09-20 12:16:24 +03:00
xs *lfn = xs_replace(fn, "/timeline/", "/local/");
link(fn, lfn);
2022-09-25 18:42:39 +03:00
snac_debug(snac, 1, xs_fmt("_timeline_write (local) %s %s", id, lfn));
2022-09-20 12:16:24 +03:00
}
2022-09-22 15:46:23 +03:00
2022-09-25 10:07:43 +03:00
if (!xs_is_null(parent)) {
2022-09-23 10:50:38 +03:00
/* update the parent, adding this id to its children list */
xs *pfn = _timeline_find_fn(snac, parent);
xs *p_msg = NULL;
if (pfn != NULL && (f = fopen(pfn, "r")) != NULL) {
xs *j;
j = xs_readall(f);
fclose(f);
p_msg = xs_json_loads(j);
}
if (p_msg == NULL)
return;
xs *meta = xs_dup(xs_dict_get(p_msg, "_snac"));
xs *children = xs_dup(xs_dict_get(meta, "children"));
2022-09-25 18:42:39 +03:00
/* add the child if it's not already there */
if (xs_list_in(children, id) == -1)
children = xs_list_append(children, id);
2022-09-23 10:50:38 +03:00
/* re-store */
meta = xs_dict_set(meta, "children", children);
p_msg = xs_dict_set(p_msg, "_snac", meta);
xs *nfn = _timeline_new_fn(snac, parent);
if ((f = fopen(nfn, "w")) != NULL) {
xs *j = xs_json_dumps_pp(p_msg, 4);
fwrite(j, strlen(j), 1, f);
fclose(f);
unlink(pfn);
2022-09-25 18:42:39 +03:00
snac_debug(snac, 1,
xs_fmt("_timeline_write updated parent %s %s", parent, nfn));
2022-09-23 10:50:38 +03:00
/* try to do the same with the local */
xs *olfn = xs_replace(pfn, "/timeline/", "/local/");
if (unlink(olfn) != -1) {
xs *nlfn = xs_replace(nfn, "/timeline/", "/local/");
link(nfn, nlfn);
2022-09-25 18:42:39 +03:00
snac_debug(snac, 1,
xs_fmt("_timeline_write updated parent (local) %s %s", parent, nlfn));
2022-09-23 10:50:38 +03:00
}
}
else
return;
/* now iterate all parents up, just renaming the files */
xs *grampa = xs_dup(xs_dict_get(meta, "parent"));
2022-09-25 10:07:43 +03:00
while (!xs_is_null(grampa)) {
2022-09-23 10:50:38 +03:00
xs *gofn = _timeline_find_fn(snac, grampa);
if (gofn == NULL)
break;
/* create the new filename */
xs *gnfn = _timeline_new_fn(snac, grampa);
rename(gofn, gnfn);
2022-09-25 23:40:31 +03:00
snac_debug(snac, 1,
2022-09-25 18:42:39 +03:00
xs_fmt("_timeline_write updated grampa %s %s", grampa, gnfn));
2022-09-23 10:50:38 +03:00
/* try to do the same with the local */
xs *golfn = xs_replace(gofn, "/timeline/", "/local/");
if (unlink(golfn) != -1) {
xs *gnlfn = xs_replace(gnfn, "/timeline/", "/local/");
link(gnfn, gnlfn);
2022-09-25 23:40:31 +03:00
snac_debug(snac, 1,
2022-09-25 18:42:39 +03:00
xs_fmt("_timeline_write updated grampa (local) %s %s", parent, gnlfn));
2022-09-23 10:50:38 +03:00
}
/* now open it and get its own parent */
if ((f = fopen(gnfn, "r")) != NULL) {
xs *j = xs_readall(f);
fclose(f);
2022-09-25 23:40:31 +03:00
xs *g_msg = xs_json_loads(j);
d_char *meta = xs_dict_get(g_msg, "_snac");
d_char *p = xs_dict_get(meta, "parent");
2022-09-23 10:50:38 +03:00
free(grampa);
2022-09-25 23:40:31 +03:00
grampa = xs_dup(p);
2022-09-23 10:50:38 +03:00
}
}
}
2022-09-20 12:16:24 +03:00
}
2022-09-20 12:31:56 +03:00
int timeline_add(snac *snac, char *id, char *o_msg, char *parent, char *referrer)
2022-09-25 18:42:39 +03:00
/* adds a message to the timeline */
{
xs *pfn = _timeline_find_fn(snac, id);
if (pfn != NULL) {
snac_log(snac, xs_fmt("timeline_add refusing rewrite %s %s", id, pfn));
2022-09-25 19:50:53 +03:00
return 0;
2022-09-25 18:42:39 +03:00
}
xs *msg = xs_dup(o_msg);
xs *md;
/* add new metadata */
md = xs_json_loads("{"
"\"children\": [],"
"\"liked_by\": [],"
"\"announced_by\": [],"
"\"version\": \"snac/2.x\","
"\"referrer\": null,"
2022-09-25 18:42:39 +03:00
"\"parent\": null"
"}");
if (!xs_is_null(parent))
md = xs_dict_set(md, "parent", parent);
if (!xs_is_null(referrer))
md = xs_dict_set(md, "parent", referrer);
2022-09-25 18:42:39 +03:00
msg = xs_dict_set(msg, "_snac", md);
_timeline_write(snac, id, msg, parent, referrer);
2022-09-25 19:28:15 +03:00
snac_log(snac, xs_fmt("timeline_add %s", id));
2022-09-25 19:50:53 +03:00
return 1;
2022-09-25 18:42:39 +03:00
}
void timeline_admire(snac *snac, char *id, char *admirer, int like)
/* updates a timeline entry with a new admiration */
{
xs *ofn = _timeline_find_fn(snac, id);
FILE *f;
if (ofn != NULL && (f = fopen(ofn, "r")) != NULL) {
xs *j1 = xs_readall(f);
fclose(f);
xs *msg = xs_json_loads(j1);
xs *meta = xs_dup(xs_dict_get(msg, "_snac"));
xs *list;
if (like)
list = xs_dup(xs_dict_get(meta, "liked_by"));
else
list = xs_dup(xs_dict_get(meta, "announced_by"));
/* add the admirer if it's not already there */
if (xs_list_in(list, admirer) == -1)
2022-09-25 18:42:39 +03:00
list = xs_list_append(list, admirer);
/* set the admirer as the referrer */
meta = xs_dict_set(meta, "referrer", admirer);
/* re-store */
if (like)
meta = xs_dict_set(meta, "liked_by", list);
else
meta = xs_dict_set(meta, "announced_by", list);
2022-09-25 18:42:39 +03:00
msg = xs_dict_set(msg, "_snac", meta);
2022-09-25 18:42:39 +03:00
unlink(ofn);
2022-09-25 18:42:39 +03:00
_timeline_write(snac, id, msg, xs_dict_get(meta, "parent"), admirer);
2022-09-25 19:28:15 +03:00
snac_log(snac, xs_fmt("timeline_admire (%s) %s %s",
like ? "Like" : "Announce", id, admirer));
2022-09-25 18:42:39 +03:00
}
else
snac_log(snac, xs_fmt("timeline_admire ignored for unknown object %s", id));
2022-09-25 18:42:39 +03:00
}
2022-09-20 12:31:56 +03:00
d_char *_following_fn(snac *snac, char *actor)
{
xs *md5 = xs_md5_hex(actor, strlen(actor));
return xs_fmt("%s/following/%s.json", snac->basedir, md5);
}
int following_add(snac *snac, char *actor, char *msg)
/* adds to the following list */
{
int ret = 201; /* created */
xs *fn = _following_fn(snac, actor);
FILE *f;
if ((f = fopen(fn, "w")) != NULL) {
xs *j = xs_json_dumps_pp(msg, 4);
fwrite(j, 1, strlen(j), f);
fclose(f);
}
else
ret = 500;
snac_debug(snac, 2, xs_fmt("following_add %s %s", actor, fn));
return ret;
}
int following_del(snac *snac, char *actor)
/* someone is no longer following us */
{
xs *fn = _following_fn(snac, actor);
unlink(fn);
snac_debug(snac, 2, xs_fmt("following_del %s %s", actor, fn));
return 200;
}
int following_check(snac *snac, char *actor)
/* checks if someone is following us */
{
xs *fn = _following_fn(snac, actor);
return !!(mtime(fn) != 0.0);
}
2022-09-20 12:38:18 +03:00
d_char *_muted_fn(snac *snac, char *actor)
{
xs *md5 = xs_md5_hex(actor, strlen(actor));
return xs_fmt("%s/muted/%s.json", snac->basedir, md5);
}
void mute(snac *snac, char *actor)
/* mutes a moron */
{
xs *fn = _muted_fn(snac, actor);
FILE *f;
if ((f = fopen(fn, "w")) != NULL) {
fprintf(f, "%s\n", actor);
fclose(f);
snac_debug(snac, 2, xs_fmt("muted %s %s", actor, fn));
}
}
void unmute(snac *snac, char *actor)
/* actor is no longer a moron */
{
xs *fn = _muted_fn(snac, actor);
unlink(fn);
snac_debug(snac, 2, xs_fmt("unmuted %s %s", actor, fn));
}
int is_muted(snac *snac, char *actor)
/* check if someone is muted */
{
xs *fn = _muted_fn(snac, actor);
return !!(mtime(fn) != 0.0);
}
2022-09-20 13:00:13 +03:00
2022-09-22 19:50:39 +03:00
d_char *_actor_fn(snac *snac, char *actor)
/* returns the file name for an actor */
{
xs *md5 = xs_md5_hex(actor, strlen(actor));
return xs_fmt("%s/actors/%s.json", snac->basedir, md5);
}
int actor_add(snac *snac, char *actor, char *msg)
/* adds a follower */
{
int ret = 201; /* created */
xs *fn = _actor_fn(snac, actor);
FILE *f;
if ((f = fopen(fn, "w")) != NULL) {
xs *j = xs_json_dumps_pp(msg, 4);
fwrite(j, 1, strlen(j), f);
fclose(f);
}
else
ret = 500;
snac_debug(snac, 2, xs_fmt("actor_add %s %s", actor, fn));
return ret;
}
2022-09-22 19:52:42 +03:00
int actor_get(snac *snac, char *actor, d_char **data)
2022-09-22 19:50:39 +03:00
/* returns an already downloaded actor */
{
xs *fn = _actor_fn(snac, actor);
float t;
float max_time;
int status;
FILE *f;
t = mtime(fn);
/* no mtime? there is nothing here */
if (t == 0.0)
return 404;
/* maximum time for the actor data to be considered stale */
max_time = 3600.0 * 36.0;
if (t + max_time < (float) time(NULL)) {
/* actor data exists but also stinks */
if ((f = fopen(fn, "a")) != NULL) {
/* write a blank at the end to 'touch' the file */
fwrite(" ", 1, 1, f);
fclose(f);
}
status = 205; /* "205: Reset Content" "110: Response Is Stale" */
2022-09-22 19:50:39 +03:00
}
else {
/* it's still valid */
status = 200;
}
if ((f = fopen(fn, "r")) != NULL) {
xs *j = xs_readall(f);
fclose(f);
*data = xs_json_loads(j);
}
else
status = 500;
return status;
}
void enqueue_input(snac *snac, char *msg, char *req)
2022-09-24 00:09:09 +03:00
/* enqueues an input message */
{
xs *ntid = tid(0);
xs *fn = xs_fmt("%s/queue/%s.json", snac->basedir, ntid);
xs *tfn = xs_fmt("%s.tmp", fn);
2022-09-24 00:09:09 +03:00
FILE *f;
if ((f = fopen(tfn, "w")) != NULL) {
xs *qmsg = xs_dict_new();
xs *j;
qmsg = xs_dict_append(qmsg, "type", "input");
qmsg = xs_dict_append(qmsg, "object", msg);
qmsg = xs_dict_append(qmsg, "req", req);
2022-09-24 00:09:09 +03:00
j = xs_json_dumps_pp(qmsg, 4);
fwrite(j, strlen(j), 1, f);
fclose(f);
rename(tfn, fn);
2022-09-25 22:45:58 +03:00
snac_debug(snac, 1, xs_fmt("enqueue_input %s", fn));
2022-09-24 00:09:09 +03:00
}
}
void enqueue_output(snac *snac, char *msg, char *actor, int retries)
2022-09-23 20:37:01 +03:00
/* enqueues an output message for an actor */
{
if (strcmp(actor, snac->actor) == 0) {
snac_debug(snac, 1, xs_str_new("enqueue refused to myself"));
return;
}
int qrt = xs_number_get(xs_dict_get(srv_config, "query_retry_minutes"));
xs *ntid = tid(retries * 60 * qrt);
xs *fn = xs_fmt("%s/queue/%s.json", snac->basedir, ntid);
xs *tfn = xs_fmt("%s.tmp", fn);
2022-09-23 20:37:01 +03:00
FILE *f;
if ((f = fopen(tfn, "w")) != NULL) {
xs *qmsg = xs_dict_new();
xs *rn = xs_number_new(retries);
xs *j;
qmsg = xs_dict_append(qmsg, "type", "output");
qmsg = xs_dict_append(qmsg, "actor", actor);
qmsg = xs_dict_append(qmsg, "object", msg);
qmsg = xs_dict_append(qmsg, "retries", rn);
j = xs_json_dumps_pp(qmsg, 4);
fwrite(j, strlen(j), 1, f);
fclose(f);
rename(tfn, fn);
2022-09-25 22:45:58 +03:00
snac_debug(snac, 1, xs_fmt("enqueue_output %s %s %d", actor, fn, retries));
2022-09-23 20:37:01 +03:00
}
}
2022-09-20 13:43:49 +03:00
d_char *queue(snac *snac)
/* returns a list with filenames that can be dequeued */
{
xs *spec = xs_fmt("%s/queue/" "*.json", snac->basedir);
d_char *list = xs_list_new();
glob_t globbuf;
time_t t = time(NULL);
if (glob(spec, 0, NULL, &globbuf) == 0) {
int n;
char *p;
for (n = 0; (p = globbuf.gl_pathv[n]) != NULL; n++) {
/* get the retry time from the basename */
char *bn = strrchr(p, '/');
time_t t2 = atol(bn + 1);
if (t2 > t)
snac_debug(snac, 2, xs_fmt("queue not yet time for %s", p));
else {
list = xs_list_append(list, p);
snac_debug(snac, 2, xs_fmt("queue ready for %s", p));
}
}
}
globfree(&globbuf);
return list;
}
2022-09-20 13:50:37 +03:00
d_char *dequeue(snac *snac, char *fn)
/* dequeues a message */
{
FILE *f;
d_char *obj = NULL;
if ((f = fopen(fn, "r")) != NULL) {
/* delete right now */
unlink(fn);
xs *j = xs_readall(f);
obj = xs_json_loads(j);
fclose(f);
}
return obj;
}