data.c 20 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899
  1. /* snac - A simple, minimalistic ActivityPub instance */
  2. /* copyright (c) 2022 grunfink - MIT license */
  3. #include "xs.h"
  4. #include "xs_io.h"
  5. #include "xs_json.h"
  6. #include "xs_openssl.h"
  7. #include "snac.h"
  8. #include <time.h>
  9. #include <glob.h>
  10. #include <sys/stat.h>
  11. int srv_open(char *basedir)
  12. /* opens a server */
  13. {
  14. int ret = 0;
  15. xs *cfg_file = NULL;
  16. FILE *f;
  17. d_char *error = NULL;
  18. srv_basedir = xs_str_new(basedir);
  19. if (xs_endswith(srv_basedir, "/"))
  20. srv_basedir = xs_crop(srv_basedir, 0, -1);
  21. cfg_file = xs_fmt("%s/server.json", basedir);
  22. if ((f = fopen(cfg_file, "r")) == NULL)
  23. error = xs_fmt("error opening '%s'", cfg_file);
  24. else {
  25. xs *cfg_data;
  26. /* read full config file */
  27. cfg_data = xs_readall(f);
  28. /* parse */
  29. srv_config = xs_json_loads(cfg_data);
  30. if (srv_config == NULL)
  31. error = xs_fmt("cannot parse '%s'", cfg_file);
  32. else {
  33. char *host;
  34. char *prefix;
  35. char *dbglvl;
  36. host = xs_dict_get(srv_config, "host");
  37. prefix = xs_dict_get(srv_config, "prefix");
  38. dbglvl = xs_dict_get(srv_config, "dbglevel");
  39. if (host == NULL || prefix == NULL)
  40. error = xs_str_new("cannot get server data");
  41. else {
  42. srv_baseurl = xs_fmt("https://%s%s", host, prefix);
  43. dbglevel = (int) xs_number_get(dbglvl);
  44. if ((dbglvl = getenv("DEBUG")) != NULL) {
  45. dbglevel = atoi(dbglvl);
  46. error = xs_fmt("DEBUG level set to %d from environment", dbglevel);
  47. }
  48. ret = 1;
  49. }
  50. }
  51. }
  52. if (ret == 0 && error != NULL)
  53. srv_log(error);
  54. return ret;
  55. }
  56. void user_free(snac *snac)
  57. /* frees a user snac */
  58. {
  59. free(snac->uid);
  60. free(snac->basedir);
  61. free(snac->config);
  62. free(snac->key);
  63. free(snac->actor);
  64. }
  65. int user_open(snac *snac, char *uid)
  66. /* opens a user */
  67. {
  68. int ret = 0;
  69. memset(snac, '\0', sizeof(struct _snac));
  70. if (validate_uid(uid)) {
  71. xs *cfg_file;
  72. FILE *f;
  73. snac->uid = xs_str_new(uid);
  74. snac->basedir = xs_fmt("%s/user/%s", srv_basedir, uid);
  75. cfg_file = xs_fmt("%s/user.json", snac->basedir);
  76. if ((f = fopen(cfg_file, "r")) != NULL) {
  77. xs *cfg_data;
  78. /* read full config file */
  79. cfg_data = xs_readall(f);
  80. fclose(f);
  81. if ((snac->config = xs_json_loads(cfg_data)) != NULL) {
  82. xs *key_file = xs_fmt("%s/key.json", snac->basedir);
  83. if ((f = fopen(key_file, "r")) != NULL) {
  84. xs *key_data;
  85. key_data = xs_readall(f);
  86. fclose(f);
  87. if ((snac->key = xs_json_loads(key_data)) != NULL) {
  88. snac->actor = xs_fmt("%s/%s", srv_baseurl, uid);
  89. ret = 1;
  90. }
  91. else
  92. srv_log(xs_fmt("cannot parse '%s'", key_file));
  93. }
  94. else
  95. srv_log(xs_fmt("error opening '%s'", key_file));
  96. }
  97. else
  98. srv_log(xs_fmt("cannot parse '%s'", cfg_file));
  99. }
  100. else
  101. srv_debug(2, xs_fmt("error opening '%s'", cfg_file));
  102. }
  103. else
  104. srv_log(xs_fmt("invalid user '%s'", uid));
  105. if (!ret)
  106. user_free(snac);
  107. return ret;
  108. }
  109. d_char *user_list(void)
  110. /* returns the list of user ids */
  111. {
  112. d_char *list;
  113. xs *spec;
  114. glob_t globbuf;
  115. globbuf.gl_offs = 1;
  116. list = xs_list_new();
  117. spec = xs_fmt("%s/user/" "*", srv_basedir);
  118. if (glob(spec, 0, NULL, &globbuf) == 0) {
  119. int n;
  120. char *p;
  121. for (n = 0; (p = globbuf.gl_pathv[n]) != NULL; n++) {
  122. if ((p = strrchr(p, '/')) != NULL)
  123. list = xs_list_append(list, p + 1);
  124. }
  125. }
  126. globfree(&globbuf);
  127. return list;
  128. }
  129. float mtime(char *fn)
  130. /* returns the mtime of a file or directory, or 0.0 */
  131. {
  132. struct stat st;
  133. float r = 0.0;
  134. if (stat(fn, &st) != -1)
  135. r = (float)st.st_mtim.tv_sec;
  136. return r;
  137. }
  138. d_char *_follower_fn(snac *snac, char *actor)
  139. {
  140. xs *md5 = xs_md5_hex(actor, strlen(actor));
  141. return xs_fmt("%s/followers/%s.json", snac->basedir, md5);
  142. }
  143. int follower_add(snac *snac, char *actor, char *msg)
  144. /* adds a follower */
  145. {
  146. int ret = 201; /* created */
  147. xs *fn = _follower_fn(snac, actor);
  148. FILE *f;
  149. if ((f = fopen(fn, "w")) != NULL) {
  150. xs *j = xs_json_dumps_pp(msg, 4);
  151. fwrite(j, 1, strlen(j), f);
  152. fclose(f);
  153. }
  154. else
  155. ret = 500;
  156. snac_debug(snac, 2, xs_fmt("follower_add %s %s", actor, fn));
  157. return ret;
  158. }
  159. int follower_del(snac *snac, char *actor)
  160. /* deletes a follower */
  161. {
  162. xs *fn = _follower_fn(snac, actor);
  163. unlink(fn);
  164. snac_debug(snac, 2, xs_fmt("follower_del %s %s", actor, fn));
  165. return 200;
  166. }
  167. int follower_check(snac *snac, char *actor)
  168. /* checks if someone is a follower */
  169. {
  170. xs *fn = _follower_fn(snac, actor);
  171. return !!(mtime(fn) != 0.0);
  172. }
  173. d_char *follower_list(snac *snac)
  174. /* returns the list of followers */
  175. {
  176. d_char *list;
  177. xs *spec;
  178. glob_t globbuf;
  179. list = xs_list_new();
  180. spec = xs_fmt("%s/followers/" "*.json", snac->basedir);
  181. if (glob(spec, 0, NULL, &globbuf) == 0) {
  182. int n;
  183. char *fn;
  184. for (n = 0; (fn = globbuf.gl_pathv[n]) != NULL; n++) {
  185. FILE *f;
  186. if ((f = fopen(fn, "r")) != NULL) {
  187. xs *j = xs_readall(f);
  188. xs *o = xs_json_loads(j);
  189. if (o != NULL)
  190. list = xs_list_append(list, o);
  191. fclose(f);
  192. }
  193. }
  194. }
  195. globfree(&globbuf);
  196. return list;
  197. }
  198. d_char *_timeline_find_fn(snac *snac, char *id)
  199. /* returns the file name of a timeline entry by its id */
  200. {
  201. xs *md5 = xs_md5_hex(id, strlen(id));
  202. xs *spec = xs_fmt("%s/timeline/" "*-%s.json", snac->basedir, md5);
  203. glob_t globbuf;
  204. d_char *fn = NULL;
  205. if (glob(spec, 0, NULL, &globbuf) == 0 && globbuf.gl_pathc) {
  206. /* get just the first file */
  207. fn = xs_str_new(globbuf.gl_pathv[0]);
  208. }
  209. globfree(&globbuf);
  210. return fn;
  211. }
  212. int timeline_here(snac *snac, char *id)
  213. /* checks if an object is already downloaded */
  214. {
  215. xs *fn = _timeline_find_fn(snac, id);
  216. return fn != NULL;
  217. }
  218. d_char *timeline_find(snac *snac, char *id)
  219. /* gets a message from the timeline by id */
  220. {
  221. xs *fn = _timeline_find_fn(snac, id);
  222. xs *msg = NULL;
  223. if (fn != NULL) {
  224. FILE *f;
  225. if ((f = fopen(fn, "r")) != NULL) {
  226. xs *j = xs_readall(f);
  227. msg = xs_json_loads(j);
  228. fclose(f);
  229. }
  230. }
  231. return msg;
  232. }
  233. void timeline_del(snac *snac, char *id)
  234. /* deletes a message from the timeline */
  235. {
  236. xs *fn = _timeline_find_fn(snac, id);
  237. if (fn != NULL) {
  238. xs *lfn = NULL;
  239. unlink(fn);
  240. snac_debug(snac, 1, xs_fmt("timeline_del %s", id));
  241. /* try to delete also from the local timeline */
  242. lfn = xs_replace(fn, "/timeline/", "/local/");
  243. if (unlink(lfn) != -1)
  244. snac_debug(snac, 1, xs_fmt("timeline_del (local) %s", id));
  245. }
  246. }
  247. d_char *timeline_get(snac *snac, char *fn)
  248. /* gets a timeline entry by file name */
  249. {
  250. d_char *d = NULL;
  251. FILE *f;
  252. if ((f = fopen(fn, "r")) != NULL) {
  253. xs *j = xs_readall(f);
  254. d = xs_json_loads(j);
  255. fclose(f);
  256. }
  257. return d;
  258. }
  259. d_char *timeline_list(snac *snac)
  260. /* returns a list of the timeline filenames */
  261. {
  262. d_char *list;
  263. xs *spec = xs_fmt("%s/timeline/" "*.json", snac->basedir);
  264. glob_t globbuf;
  265. int max;
  266. /* maximum number of items in the timeline */
  267. max = xs_number_get(xs_dict_get(srv_config, "max_timeline_entries"));
  268. list = xs_list_new();
  269. /* get the list in reverse order */
  270. if (glob(spec, 0, NULL, &globbuf) == 0) {
  271. int n;
  272. if (max > globbuf.gl_pathc)
  273. max = globbuf.gl_pathc;
  274. for (n = 0; n < max; n++) {
  275. char *fn = globbuf.gl_pathv[globbuf.gl_pathc - n - 1];
  276. list = xs_list_append(list, fn);
  277. }
  278. }
  279. globfree(&globbuf);
  280. return list;
  281. }
  282. d_char *_timeline_new_fn(snac *snac, char *id)
  283. /* creates a new filename */
  284. {
  285. xs *ntid = tid(0);
  286. xs *md5 = xs_md5_hex(id, strlen(id));
  287. return xs_fmt("%s/timeline/%s-%s.json", snac->basedir, ntid, md5);
  288. }
  289. void _timeline_write(snac *snac, char *id, char *msg, char *parent)
  290. /* writes a timeline entry and refreshes the ancestors */
  291. {
  292. xs *fn = _timeline_new_fn(snac, id);
  293. FILE *f;
  294. if ((f = fopen(fn, "w")) != NULL) {
  295. xs *j = xs_json_dumps_pp(msg, 4);
  296. fwrite(j, strlen(j), 1, f);
  297. fclose(f);
  298. snac_debug(snac, 1, xs_fmt("_timeline_write %s %s", id, fn));
  299. }
  300. /* related to this user? link to local timeline */
  301. if (xs_startswith(id, snac->actor) ||
  302. (!xs_is_null(parent) && xs_startswith(parent, snac->actor))) {
  303. xs *lfn = xs_replace(fn, "/timeline/", "/local/");
  304. link(fn, lfn);
  305. snac_debug(snac, 1, xs_fmt("_timeline_write (local) %s %s", id, lfn));
  306. }
  307. if (!xs_is_null(parent)) {
  308. /* update the parent, adding this id to its children list */
  309. xs *pfn = _timeline_find_fn(snac, parent);
  310. xs *p_msg = NULL;
  311. if (pfn != NULL && (f = fopen(pfn, "r")) != NULL) {
  312. xs *j;
  313. j = xs_readall(f);
  314. fclose(f);
  315. p_msg = xs_json_loads(j);
  316. }
  317. if (p_msg == NULL)
  318. return;
  319. xs *meta = xs_dup(xs_dict_get(p_msg, "_snac"));
  320. xs *children = xs_dup(xs_dict_get(meta, "children"));
  321. /* add the child if it's not already there */
  322. if (xs_list_in(children, id) == -1)
  323. children = xs_list_append(children, id);
  324. /* re-store */
  325. meta = xs_dict_set(meta, "children", children);
  326. p_msg = xs_dict_set(p_msg, "_snac", meta);
  327. xs *nfn = _timeline_new_fn(snac, parent);
  328. if ((f = fopen(nfn, "w")) != NULL) {
  329. xs *j = xs_json_dumps_pp(p_msg, 4);
  330. fwrite(j, strlen(j), 1, f);
  331. fclose(f);
  332. unlink(pfn);
  333. snac_debug(snac, 1,
  334. xs_fmt("_timeline_write updated parent %s %s", parent, nfn));
  335. /* try to do the same with the local */
  336. xs *olfn = xs_replace(pfn, "/timeline/", "/local/");
  337. if (unlink(olfn) != -1) {
  338. xs *nlfn = xs_replace(nfn, "/timeline/", "/local/");
  339. link(nfn, nlfn);
  340. snac_debug(snac, 1,
  341. xs_fmt("_timeline_write updated parent (local) %s %s", parent, nlfn));
  342. }
  343. }
  344. else
  345. return;
  346. /* now iterate all parents up, just renaming the files */
  347. xs *grampa = xs_dup(xs_dict_get(meta, "parent"));
  348. while (!xs_is_null(grampa)) {
  349. xs *gofn = _timeline_find_fn(snac, grampa);
  350. if (gofn == NULL)
  351. break;
  352. /* create the new filename */
  353. xs *gnfn = _timeline_new_fn(snac, grampa);
  354. rename(gofn, gnfn);
  355. snac_debug(snac, 2,
  356. xs_fmt("_timeline_write updated grampa %s %s", grampa, gnfn));
  357. /* try to do the same with the local */
  358. xs *golfn = xs_replace(gofn, "/timeline/", "/local/");
  359. if (unlink(golfn) != -1) {
  360. xs *gnlfn = xs_replace(gnfn, "/timeline/", "/local/");
  361. link(gnfn, gnlfn);
  362. snac_debug(snac, 2,
  363. xs_fmt("_timeline_write updated grampa (local) %s %s", parent, gnlfn));
  364. }
  365. /* now open it and get its own parent */
  366. if ((f = fopen(gnfn, "r")) != NULL) {
  367. xs *j = xs_readall(f);
  368. fclose(f);
  369. xs *g_msg = xs_json_loads(j);
  370. xs *meta = xs_dict_get(g_msg, "_snac");
  371. d_char *p = xs_dict_get(meta, "parent");
  372. free(grampa);
  373. if (!xs_is_null(p))
  374. p = xs_dup(p);
  375. grampa = p;
  376. }
  377. }
  378. }
  379. }
  380. int timeline_add(snac *snac, char *id, char *o_msg, char *parent)
  381. /* adds a message to the timeline */
  382. {
  383. xs *pfn = _timeline_find_fn(snac, id);
  384. if (pfn != NULL) {
  385. snac_log(snac, xs_fmt("timeline_add refusing rewrite %s %s", id, pfn));
  386. return 0;
  387. }
  388. xs *msg = xs_dup(o_msg);
  389. xs *md;
  390. /* add new metadata */
  391. md = xs_json_loads("{"
  392. "\"children\": [],"
  393. "\"liked_by\": [],"
  394. "\"announced_by\": [],"
  395. "\"version\": \"snac/2.x\","
  396. "\"parent\": null"
  397. "}");
  398. if (!xs_is_null(parent))
  399. md = xs_dict_set(md, "parent", parent);
  400. msg = xs_dict_set(msg, "_snac", md);
  401. _timeline_write(snac, id, msg, parent);
  402. snac_log(snac, xs_fmt("timeline_add %s", id));
  403. return 1;
  404. }
  405. void timeline_admire(snac *snac, char *id, char *admirer, int like)
  406. /* updates a timeline entry with a new admiration */
  407. {
  408. xs *ofn = _timeline_find_fn(snac, id);
  409. FILE *f;
  410. if (ofn != NULL && (f = fopen(ofn, "r")) != NULL) {
  411. int changed = 0;
  412. xs *j1 = xs_readall(f);
  413. fclose(f);
  414. xs *msg = xs_json_loads(j1);
  415. xs *meta = xs_dup(xs_dict_get(msg, "_snac"));
  416. xs *list;
  417. if (like)
  418. list = xs_dup(xs_dict_get(meta, "liked_by"));
  419. else
  420. list = xs_dup(xs_dict_get(meta, "announced_by"));
  421. /* add the admirer if it's not already there */
  422. if (xs_list_in(list, admirer) == -1) {
  423. list = xs_list_append(list, admirer);
  424. changed = 1;
  425. }
  426. if (changed) {
  427. /* re-store */
  428. if (like)
  429. meta = xs_dict_set(meta, "liked_by", list);
  430. else
  431. meta = xs_dict_set(meta, "announced_by", list);
  432. msg = xs_dict_set(msg, "_snac", meta);
  433. unlink(ofn);
  434. _timeline_write(snac, id, msg, xs_dict_get(meta, "parent"));
  435. snac_log(snac, xs_fmt("timeline_admire (%s) %s %s",
  436. like ? "Like" : "Announce", id, admirer));
  437. }
  438. }
  439. }
  440. d_char *_following_fn(snac *snac, char *actor)
  441. {
  442. xs *md5 = xs_md5_hex(actor, strlen(actor));
  443. return xs_fmt("%s/following/%s.json", snac->basedir, md5);
  444. }
  445. int following_add(snac *snac, char *actor, char *msg)
  446. /* adds to the following list */
  447. {
  448. int ret = 201; /* created */
  449. xs *fn = _following_fn(snac, actor);
  450. FILE *f;
  451. if ((f = fopen(fn, "w")) != NULL) {
  452. xs *j = xs_json_dumps_pp(msg, 4);
  453. fwrite(j, 1, strlen(j), f);
  454. fclose(f);
  455. }
  456. else
  457. ret = 500;
  458. snac_debug(snac, 2, xs_fmt("following_add %s %s", actor, fn));
  459. return ret;
  460. }
  461. int following_del(snac *snac, char *actor)
  462. /* someone is no longer following us */
  463. {
  464. xs *fn = _following_fn(snac, actor);
  465. unlink(fn);
  466. snac_debug(snac, 2, xs_fmt("following_del %s %s", actor, fn));
  467. return 200;
  468. }
  469. int following_check(snac *snac, char *actor)
  470. /* checks if someone is following us */
  471. {
  472. xs *fn = _following_fn(snac, actor);
  473. return !!(mtime(fn) != 0.0);
  474. }
  475. d_char *_muted_fn(snac *snac, char *actor)
  476. {
  477. xs *md5 = xs_md5_hex(actor, strlen(actor));
  478. return xs_fmt("%s/muted/%s.json", snac->basedir, md5);
  479. }
  480. void mute(snac *snac, char *actor)
  481. /* mutes a moron */
  482. {
  483. xs *fn = _muted_fn(snac, actor);
  484. FILE *f;
  485. if ((f = fopen(fn, "w")) != NULL) {
  486. fprintf(f, "%s\n", actor);
  487. fclose(f);
  488. snac_debug(snac, 2, xs_fmt("muted %s %s", actor, fn));
  489. }
  490. }
  491. void unmute(snac *snac, char *actor)
  492. /* actor is no longer a moron */
  493. {
  494. xs *fn = _muted_fn(snac, actor);
  495. unlink(fn);
  496. snac_debug(snac, 2, xs_fmt("unmuted %s %s", actor, fn));
  497. }
  498. int is_muted(snac *snac, char *actor)
  499. /* check if someone is muted */
  500. {
  501. xs *fn = _muted_fn(snac, actor);
  502. return !!(mtime(fn) != 0.0);
  503. }
  504. d_char *_actor_fn(snac *snac, char *actor)
  505. /* returns the file name for an actor */
  506. {
  507. xs *md5 = xs_md5_hex(actor, strlen(actor));
  508. return xs_fmt("%s/actors/%s.json", snac->basedir, md5);
  509. }
  510. int actor_add(snac *snac, char *actor, char *msg)
  511. /* adds a follower */
  512. {
  513. int ret = 201; /* created */
  514. xs *fn = _actor_fn(snac, actor);
  515. FILE *f;
  516. if ((f = fopen(fn, "w")) != NULL) {
  517. xs *j = xs_json_dumps_pp(msg, 4);
  518. fwrite(j, 1, strlen(j), f);
  519. fclose(f);
  520. }
  521. else
  522. ret = 500;
  523. snac_debug(snac, 2, xs_fmt("actor_add %s %s", actor, fn));
  524. return ret;
  525. }
  526. int actor_get(snac *snac, char *actor, d_char **data)
  527. /* returns an already downloaded actor */
  528. {
  529. xs *fn = _actor_fn(snac, actor);
  530. float t;
  531. float max_time;
  532. int status;
  533. FILE *f;
  534. t = mtime(fn);
  535. /* no mtime? there is nothing here */
  536. if (t == 0.0)
  537. return 404;
  538. /* maximum time for the actor data to be considered stale */
  539. max_time = 3600.0 * 36.0;
  540. if (t + max_time < (float) time(NULL)) {
  541. /* actor data exists but also stinks */
  542. if ((f = fopen(fn, "a")) != NULL) {
  543. /* write a blank at the end to 'touch' the file */
  544. fwrite(" ", 1, 1, f);
  545. fclose(f);
  546. }
  547. status = 205; /* "205: Reset Content" "110: Response Is Stale" */
  548. }
  549. else {
  550. /* it's still valid */
  551. status = 200;
  552. }
  553. if ((f = fopen(fn, "r")) != NULL) {
  554. xs *j = xs_readall(f);
  555. fclose(f);
  556. *data = xs_json_loads(j);
  557. }
  558. else
  559. status = 500;
  560. return status;
  561. }
  562. void enqueue_input(snac *snac, char *msg, char *req)
  563. /* enqueues an input message */
  564. {
  565. xs *ntid = tid(0);
  566. xs *fn = xs_fmt("%s/queue/%s.json", snac->basedir, ntid);
  567. xs *tfn = xs_fmt("%s.tmp", fn);
  568. FILE *f;
  569. if ((f = fopen(tfn, "w")) != NULL) {
  570. xs *qmsg = xs_dict_new();
  571. xs *j;
  572. qmsg = xs_dict_append(qmsg, "type", "input");
  573. qmsg = xs_dict_append(qmsg, "object", msg);
  574. qmsg = xs_dict_append(qmsg, "req", req);
  575. j = xs_json_dumps_pp(qmsg, 4);
  576. fwrite(j, strlen(j), 1, f);
  577. fclose(f);
  578. rename(tfn, fn);
  579. snac_debug(snac, 1, xs_fmt("enqueue_input %s", fn));
  580. }
  581. }
  582. void enqueue_output(snac *snac, char *msg, char *actor, int retries)
  583. /* enqueues an output message for an actor */
  584. {
  585. if (strcmp(actor, snac->actor) == 0) {
  586. snac_debug(snac, 1, xs_str_new("enqueue refused to myself"));
  587. return;
  588. }
  589. int qrt = xs_number_get(xs_dict_get(srv_config, "query_retry_minutes"));
  590. xs *ntid = tid(retries * 60 * qrt);
  591. xs *fn = xs_fmt("%s/queue/%s.json", snac->basedir, ntid);
  592. xs *tfn = xs_fmt("%s.tmp", fn);
  593. FILE *f;
  594. if ((f = fopen(tfn, "w")) != NULL) {
  595. xs *qmsg = xs_dict_new();
  596. xs *rn = xs_number_new(retries);
  597. xs *j;
  598. qmsg = xs_dict_append(qmsg, "type", "output");
  599. qmsg = xs_dict_append(qmsg, "actor", actor);
  600. qmsg = xs_dict_append(qmsg, "object", msg);
  601. qmsg = xs_dict_append(qmsg, "retries", rn);
  602. j = xs_json_dumps_pp(qmsg, 4);
  603. fwrite(j, strlen(j), 1, f);
  604. fclose(f);
  605. rename(tfn, fn);
  606. snac_debug(snac, 1, xs_fmt("enqueue_output %s %s %d", actor, fn, retries));
  607. }
  608. }
  609. d_char *queue(snac *snac)
  610. /* returns a list with filenames that can be dequeued */
  611. {
  612. xs *spec = xs_fmt("%s/queue/" "*.json", snac->basedir);
  613. d_char *list = xs_list_new();
  614. glob_t globbuf;
  615. time_t t = time(NULL);
  616. if (glob(spec, 0, NULL, &globbuf) == 0) {
  617. int n;
  618. char *p;
  619. for (n = 0; (p = globbuf.gl_pathv[n]) != NULL; n++) {
  620. /* get the retry time from the basename */
  621. char *bn = strrchr(p, '/');
  622. time_t t2 = atol(bn + 1);
  623. if (t2 > t)
  624. snac_debug(snac, 2, xs_fmt("queue not yet time for %s", p));
  625. else {
  626. list = xs_list_append(list, p);
  627. snac_debug(snac, 2, xs_fmt("queue ready for %s", p));
  628. }
  629. }
  630. }
  631. globfree(&globbuf);
  632. return list;
  633. }
  634. d_char *dequeue(snac *snac, char *fn)
  635. /* dequeues a message */
  636. {
  637. FILE *f;
  638. d_char *obj = NULL;
  639. if ((f = fopen(fn, "r")) != NULL) {
  640. /* delete right now */
  641. unlink(fn);
  642. xs *j = xs_readall(f);
  643. obj = xs_json_loads(j);
  644. fclose(f);
  645. }
  646. return obj;
  647. }