697b63b47b986111ca86f2e88609010b90e09f9b
[citadel.git] / citadel / modules / rssclient / serv_rssclient.c
1 /*
2  * Bring external RSS feeds into rooms.
3  *
4  * Copyright (c) 2007-2015 by the citadel.org team
5  *
6  * This program is open source software; you can redistribute it and/or modify
7  * it under the terms of the GNU General Public License version 3.
8  *
9  * This program is distributed in the hope that it will be useful,
10  * but WITHOUT ANY WARRANTY; without even the implied warranty of
11  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.See the
12  * GNU General Public License for more details.
13  */
14
15 #include <stdlib.h>
16 #include <unistd.h>
17 #include <stdio.h>
18
19 #if TIME_WITH_SYS_TIME
20 # include <sys/time.h>
21 # include <time.h>
22 #else
23 # if HAVE_SYS_TIME_H
24 #include <sys/time.h>
25 # else
26 #include <time.h>
27 # endif
28 #endif
29
30 #include <ctype.h>
31 #include <string.h>
32 #include <errno.h>
33 #include <sys/types.h>
34 #include <sys/stat.h>
35 #include <expat.h>
36 #include <curl/curl.h>
37 #include <libcitadel.h>
38 #include "citadel.h"
39 #include "server.h"
40 #include "citserver.h"
41 #include "support.h"
42 #include "config.h"
43 #include "threads.h"
44 #include "ctdl_module.h"
45 #include "msgbase.h"
46 #include "parsedate.h"
47 #include "database.h"
48 #include "citadel_dirs.h"
49 #include "md5.h"
50 #include "context.h"
51 #include "event_client.h"
52 #include "rss_atom_parser.h"
53
54
55 #define TMP_MSGDATA 0xFF
56 #define TMP_SHORTER_URL_OFFSET 0xFE
57 #define TMP_SHORTER_URLS 0xFD
58
59 time_t last_run = 0L;
60
61 pthread_mutex_t RSSQueueMutex; /* locks the access to the following vars: */
62 HashList *RSSQueueRooms = NULL; /* rss_room_counter */
63 HashList *RSSFetchUrls = NULL; /*->rss_aggregator;->RefCount access locked*/
64
65 eNextState RSSAggregator_Terminate(AsyncIO *IO);
66 eNextState RSSAggregator_TerminateDB(AsyncIO *IO);
67 eNextState RSSAggregator_ShutdownAbort(AsyncIO *IO);
68 struct CitContext rss_CC;
69
70 struct rssnetcfg *rnclist = NULL;
71 int RSSClientDebugEnabled = 0;
72 #define N ((rss_aggregator*)IO->Data)->Cfg.QRnumber
73
74 #define DBGLOG(LEVEL) if ((LEVEL != LOG_DEBUG) || (RSSClientDebugEnabled != 0))
75
76 #define EVRSSC_syslog(LEVEL, FORMAT, ...)                               \
77         DBGLOG(LEVEL) syslog(LEVEL,                                     \
78                              "%s[%ld]CC[%d][%ld]RSS" FORMAT,            \
79                              IOSTR, IO->ID, CCID, N, __VA_ARGS__)
80
81 #define EVRSSCM_syslog(LEVEL, FORMAT)                                   \
82         DBGLOG(LEVEL) syslog(LEVEL,                                     \
83                              "%s[%ld]CC[%d][%ld]RSS" FORMAT,            \
84                              IOSTR, IO->ID, CCID, N)
85
86 #define EVRSSQ_syslog(LEVEL, FORMAT, ...)                               \
87         DBGLOG(LEVEL) syslog(LEVEL, "RSS" FORMAT,                       \
88                              __VA_ARGS__)
89 #define EVRSSQM_syslog(LEVEL, FORMAT)                   \
90         DBGLOG(LEVEL) syslog(LEVEL, "RSS" FORMAT)
91
92 #define EVRSSCSM_syslog(LEVEL, FORMAT)                                  \
93         DBGLOG(LEVEL) syslog(LEVEL, "%s[%ld][%ld]RSS" FORMAT,           \
94                              IOSTR, IO->ID, N)
95
96 typedef enum _RSSState {
97         eRSSCreated,
98         eRSSFetching,
99         eRSSFailure,
100         eRSSParsing,
101         eRSSUT
102 } RSSState;
103 ConstStr RSSStates[] = {
104         {HKEY("Aggregator created")},
105         {HKEY("Fetching content")},
106         {HKEY("Failed")},
107         {HKEY("parsing content")},
108         {HKEY("checking usetable")}
109 };
110
111
112 static size_t GetLocationString( void *ptr, size_t size, size_t nmemb, void *userdata)
113 {
114 #define LOCATION "location"
115         if (strncasecmp((char*)ptr, LOCATION, sizeof(LOCATION) - 1) == 0)
116         {
117                 AsyncIO *IO = (AsyncIO *) userdata;
118                 rss_aggregator *RSSAggr = (rss_aggregator *)IO->Data;
119
120                 char *pch = (char*) ptr;
121                 char *pche;
122                 
123                 pche = pch + (size * nmemb);
124                 pch += sizeof(LOCATION);
125                 
126                 while (isspace(*pch) || (*pch == ':'))
127                         pch ++;
128
129                 while (isspace(*pche) || (*pche == '\0'))
130                         pche--;
131                 if (RSSAggr->RedirectUrl == NULL) {
132                         RSSAggr->RedirectUrl = NewStrBufPlain(pch, pche - pch + 1);
133                 }
134                 else {
135                         FlushStrBuf(RSSAggr->RedirectUrl);
136                         StrBufPlain(RSSAggr->RedirectUrl, pch, pche - pch + 1); 
137                 }
138         }
139         return size * nmemb;
140 }
141
142 static void SetRSSState(AsyncIO *IO, RSSState State)
143 {
144         CitContext* CCC = IO->CitContext;
145         if (CCC != NULL)
146                 memcpy(CCC->cs_clientname, RSSStates[State].Key, RSSStates[State].len + 1);
147 }
148
149 void DeleteRoomReference(long QRnumber)
150 {
151         HashPos *At;
152         long HKLen;
153         const char *HK;
154         void *vData = NULL;
155         rss_room_counter *pRoomC;
156
157         At = GetNewHashPos(RSSQueueRooms, 0);
158
159         if (GetHashPosFromKey(RSSQueueRooms, LKEY(QRnumber), At))
160         {
161                 GetHashPos(RSSQueueRooms, At, &HKLen, &HK, &vData);
162                 if (vData != NULL)
163                 {
164                         pRoomC = (rss_room_counter *) vData;
165                         pRoomC->count --;
166                         if (pRoomC->count == 0)
167                                 DeleteEntryFromHash(RSSQueueRooms, At);
168                 }
169         }
170         DeleteHashPos(&At);
171 }
172
173 void UnlinkRooms(rss_aggregator *RSSAggr)
174 {
175         DeleteRoomReference(RSSAggr->Cfg.QRnumber);
176         if (RSSAggr->OtherQRnumbers != NULL)
177         {
178                 long HKLen;
179                 const char *HK;
180                 HashPos *At;
181                 void *vData;
182
183                 At = GetNewHashPos(RSSAggr->OtherQRnumbers, 0);
184                 while (! server_shutting_down &&
185                        GetNextHashPos(RSSAggr->OtherQRnumbers,
186                                       At,
187                                       &HKLen, &HK,
188                                       &vData) &&
189                        (vData != NULL))
190                 {
191                         pRSSConfig *Data = (pRSSConfig*) vData;
192                         DeleteRoomReference(Data->QRnumber);
193                 }
194
195                 DeleteHashPos(&At);
196         }
197 }
198
199 void UnlinkRSSAggregator(rss_aggregator *RSSAggr)
200 {
201         HashPos *At;
202
203         pthread_mutex_lock(&RSSQueueMutex);
204         UnlinkRooms(RSSAggr);
205
206         At = GetNewHashPos(RSSFetchUrls, 0);
207         if (GetHashPosFromKey(RSSFetchUrls, SKEY(RSSAggr->Url), At))
208         {
209                 DeleteEntryFromHash(RSSFetchUrls, At);
210         }
211         DeleteHashPos(&At);
212         last_run = time(NULL);
213         pthread_mutex_unlock(&RSSQueueMutex);
214 }
215
216 void DeleteRssCfg(void *vptr)
217 {
218         rss_aggregator *RSSAggr = (rss_aggregator *)vptr;
219         AsyncIO *IO = &RSSAggr->IO;
220
221         if (IO->CitContext != NULL)
222                 EVRSSCM_syslog(LOG_DEBUG, "RSS: destroying\n");
223
224         FreeStrBuf(&RSSAggr->Url);
225         FreeStrBuf(&RSSAggr->RedirectUrl);
226         FreeStrBuf(&RSSAggr->rooms);
227         FreeStrBuf(&RSSAggr->CData);
228         FreeStrBuf(&RSSAggr->Key);
229         DeleteHash(&RSSAggr->OtherQRnumbers);
230
231         DeleteHashPos (&RSSAggr->Pos);
232         DeleteHash (&RSSAggr->Messages);
233         if (RSSAggr->recp.recp_room != NULL)
234                 free(RSSAggr->recp.recp_room);
235
236
237         if (RSSAggr->Item != NULL)
238         {
239                 flush_rss_item(RSSAggr->Item);
240
241                 free(RSSAggr->Item);
242         }
243
244         FreeAsyncIOContents(&RSSAggr->IO);
245         memset(RSSAggr, 0, sizeof(rss_aggregator));
246         free(RSSAggr);
247 }
248
249 eNextState RSSAggregator_Terminate(AsyncIO *IO)
250 {
251         rss_aggregator *RSSAggr = (rss_aggregator *)IO->Data;
252
253         EVRSSCM_syslog(LOG_DEBUG, "RSS: Terminating.\n");
254
255         StopCurlWatchers(IO);
256         UnlinkRSSAggregator(RSSAggr);
257         return eAbort;
258 }
259
260 eNextState RSSAggregator_TerminateDB(AsyncIO *IO)
261 {
262         rss_aggregator *RSSAggr = (rss_aggregator *)IO->Data;
263
264         EVRSSCM_syslog(LOG_DEBUG, "RSS: Terminating.\n");
265
266
267         StopDBWatchers(&RSSAggr->IO);
268         UnlinkRSSAggregator(RSSAggr);
269         return eAbort;
270 }
271
272 eNextState RSSAggregator_ShutdownAbort(AsyncIO *IO)
273 {
274         const char *pUrl;
275         rss_aggregator *RSSAggr = (rss_aggregator *)IO->Data;
276
277         pUrl = IO->ConnectMe->PlainUrl;
278         if (pUrl == NULL)
279                 pUrl = "";
280
281         EVRSSC_syslog(LOG_DEBUG, "RSS: Aborting by shutdown: %s.\n", pUrl);
282
283         StopCurlWatchers(IO);
284         UnlinkRSSAggregator(RSSAggr);
285         return eAbort;
286 }
287
288 void AppendLink(StrBuf *Message,
289                 StrBuf *link,
290                 StrBuf *LinkTitle,
291                 const char *Title)
292 {
293         if (StrLength(link) > 0)
294         {
295                 StrBufAppendBufPlain(Message, HKEY("<a href=\""), 0);
296                 StrBufAppendBuf(Message, link, 0);
297                 StrBufAppendBufPlain(Message, HKEY("\">"), 0);
298                 if (StrLength(LinkTitle) > 0)
299                         StrBufAppendBuf(Message, LinkTitle, 0);
300                 else if ((Title != NULL) && !IsEmptyStr(Title))
301                         StrBufAppendBufPlain(Message, Title, -1, 0);
302                 else
303                         StrBufAppendBuf(Message, link, 0);
304                 StrBufAppendBufPlain(Message, HKEY("</a><br>\n"), 0);
305         }
306 }
307
308
309 int rss_format_item(AsyncIO *IO, networker_save_message *SaveMsg)
310 {
311         StrBuf *Message;
312         int msglen = 0;
313
314         if (StrLength(SaveMsg->description) + 
315             StrLength(SaveMsg->link) + 
316             StrLength(SaveMsg->linkTitle) + 
317             StrLength(SaveMsg->reLink) +
318             StrLength(SaveMsg->reLinkTitle) +
319             StrLength(SaveMsg->title) == 0)
320         {
321                 EVRSSCM_syslog(LOG_INFO, "Refusing to save empty message.");
322                 return 0;
323         }
324
325         CM_Flush(&SaveMsg->Msg);
326
327         if (SaveMsg->author_or_creator != NULL) {
328
329                 char *From;
330                 StrBuf *Encoded = NULL;
331                 int FromAt;
332
333                 From = html_to_ascii(ChrPtr(SaveMsg->author_or_creator),
334                                      StrLength(SaveMsg->author_or_creator),
335                                      512, 0);
336                 StrBufPlain(SaveMsg->author_or_creator, From, -1);
337                 StrBufTrim(SaveMsg->author_or_creator);
338                 free(From);
339
340                 FromAt = strchr(ChrPtr(SaveMsg->author_or_creator), '@') != NULL;
341                 if (!FromAt && StrLength (SaveMsg->author_email) > 0)
342                 {
343                         StrBufRFC2047encode(&Encoded, SaveMsg->author_or_creator);
344                         CM_SetAsFieldSB(&SaveMsg->Msg, eAuthor, &Encoded);
345                         CM_SetAsFieldSB(&SaveMsg->Msg, eMessagePath, &SaveMsg->author_email);
346                 }
347                 else
348                 {
349                         if (FromAt)
350                         {
351                                 CM_SetAsFieldSB(&SaveMsg->Msg, eAuthor, &SaveMsg->author_or_creator);
352                                 CM_CopyField(&SaveMsg->Msg, eMessagePath, eAuthor);
353                         }
354                         else
355                         {
356                                 StrBufRFC2047encode(&Encoded,
357                                                     SaveMsg->author_or_creator);
358                                 CM_SetAsFieldSB(&SaveMsg->Msg, eAuthor, &Encoded);
359                                 CM_SetField(&SaveMsg->Msg, eMessagePath, HKEY("rss@localhost"));
360
361                         }
362                 }
363         }
364         else {
365                 CM_SetField(&SaveMsg->Msg, eAuthor, HKEY("rss"));
366         }
367
368         CM_SetField(&SaveMsg->Msg, eNodeName, CtdlGetConfigStr("c_nodename"), strlen(CtdlGetConfigStr("c_nodename")));
369         if (SaveMsg->title != NULL) {
370                 long len;
371                 char *Sbj;
372                 StrBuf *Encoded, *QPEncoded;
373
374                 QPEncoded = NULL;
375                 StrBufSpaceToBlank(SaveMsg->title);
376                 len = StrLength(SaveMsg->title);
377                 Sbj = html_to_ascii(ChrPtr(SaveMsg->title), len, 512, 0);
378                 len = strlen(Sbj);
379                 if ((len > 0) && (Sbj[len - 1] == '\n'))
380                 {
381                         len --;
382                         Sbj[len] = '\0';
383                 }
384                 Encoded = NewStrBufPlain(Sbj, len);
385                 free(Sbj);
386
387                 StrBufTrim(Encoded);
388                 StrBufRFC2047encode(&QPEncoded, Encoded);
389
390                 CM_SetAsFieldSB(&SaveMsg->Msg, eMsgSubject, &QPEncoded);
391                 FreeStrBuf(&Encoded);
392         }
393         if (SaveMsg->link == NULL)
394                 SaveMsg->link = NewStrBufPlain(HKEY(""));
395
396 #if 0 /* temporarily disable shorter urls. */
397         SaveMsg->Msg.cm_fields[TMP_SHORTER_URLS] =
398                 GetShorterUrls(SaveMsg->description);
399 #endif
400
401         msglen += 1024 + StrLength(SaveMsg->link) + StrLength(SaveMsg->description) ;
402
403         Message = NewStrBufPlain(NULL, msglen);
404
405         StrBufPlain(Message, HKEY(
406                             "Content-type: text/html; charset=\"UTF-8\"\r\n\r\n"
407                             "<html><body>\n"));
408 #if 0 /* disable shorter url for now. */
409         SaveMsg->Msg.cm_fields[TMP_SHORTER_URL_OFFSET] = StrLength(Message);
410 #endif
411         StrBufAppendBuf(Message, SaveMsg->description, 0);
412         StrBufAppendBufPlain(Message, HKEY("<br><br>\n"), 0);
413
414         AppendLink(Message, SaveMsg->link, SaveMsg->linkTitle, NULL);
415         AppendLink(Message, SaveMsg->reLink, SaveMsg->reLinkTitle, "Reply to this");
416         StrBufAppendBufPlain(Message, HKEY("</body></html>\n"), 0);
417
418         SaveMsg->Message = Message;
419         return 1;
420 }
421
422 eNextState RSSSaveMessage(AsyncIO *IO)
423 {
424         long len;
425         const char *Key;
426         rss_aggregator *RSSAggr = (rss_aggregator *) IO->Data;
427
428         if (rss_format_item(IO, RSSAggr->ThisMsg))
429         {
430                 CM_SetAsFieldSB(&RSSAggr->ThisMsg->Msg, eMesageText,
431                                        &RSSAggr->ThisMsg->Message);
432
433                 CtdlSubmitMsg(&RSSAggr->ThisMsg->Msg, &RSSAggr->recp, NULL, 0);
434                 
435                 /* write the uidl to the use table so we don't store this item again */
436                 
437                 CheckIfAlreadySeen("RSS Item Insert", RSSAggr->ThisMsg->MsgGUID, EvGetNow(IO), 0, eWrite, CCID, IO->ID);
438         }
439
440         if (GetNextHashPos(RSSAggr->Messages,
441                            RSSAggr->Pos,
442                            &len, &Key,
443                            (void**) &RSSAggr->ThisMsg))
444                 return NextDBOperation(IO, RSS_FetchNetworkUsetableEntry);
445         else
446                 return eAbort;
447 }
448
449 eNextState RSS_FetchNetworkUsetableEntry(AsyncIO *IO)
450 {
451         static const time_t antiExpire = USETABLE_ANTIEXPIRE_HIRES;
452 #ifndef DEBUG_RSS
453         time_t seenstamp = 0;
454         const char *Key;
455         long len;
456         rss_aggregator *Ctx = (rss_aggregator *) IO->Data;
457
458         /* Find out if we've already seen this item */
459 // todo: expiry?
460         SetRSSState(IO, eRSSUT);
461         seenstamp = CheckIfAlreadySeen("RSS Item Seen",
462                                        Ctx->ThisMsg->MsgGUID,
463                                        EvGetNow(IO),
464                                        antiExpire,
465                                        eCheckUpdate,
466                                        CCID, IO->ID);
467         if (seenstamp != 0)
468         {
469                 /* Item has already been seen */
470                 EVRSSC_syslog(LOG_DEBUG,
471                               "%s has already been seen - %ld < %ld",
472                               ChrPtr(Ctx->ThisMsg->MsgGUID),
473                               seenstamp, antiExpire);
474
475                 SetRSSState(IO, eRSSParsing);
476
477                 if (GetNextHashPos(Ctx->Messages,
478                                    Ctx->Pos,
479                                    &len, &Key,
480                                    (void**) &Ctx->ThisMsg))
481                         return NextDBOperation(
482                                 IO,
483                                 RSS_FetchNetworkUsetableEntry);
484                 else
485                         return eAbort;
486         }
487         else
488 #endif
489         {
490                 /* Item has already been seen */
491                 EVRSSC_syslog(LOG_DEBUG,
492                               "%s Parsing - %ld >= %ld",
493                               ChrPtr(Ctx->ThisMsg->MsgGUID),
494                               seenstamp, antiExpire);
495                 SetRSSState(IO, eRSSParsing);
496
497                 NextDBOperation(IO, RSSSaveMessage);
498                 return eSendMore;
499         }
500         return eSendMore;
501 }
502
503 void UpdateLastKnownGood(pRSSConfig *pCfg, time_t now)
504 {
505         OneRoomNetCfg* pRNCfg;
506         begin_critical_section(S_NETCONFIGS);
507         pRNCfg = CtdlGetNetCfgForRoom (pCfg->QRnumber);
508         if (pRNCfg != NULL)
509         {
510                 RSSCfgLine *RSSCfg = (RSSCfgLine *)pRNCfg->NetConfigs[rssclient];
511
512                 while (RSSCfg != NULL)
513                 {
514                         if (RSSCfg == pCfg->pCfg)
515                                 break;
516
517                         RSSCfg = RSSCfg->next;
518                 }
519                 if (RSSCfg != NULL)
520                 {
521                         pRNCfg->changed = 1;
522                         RSSCfg->last_known_good = now;
523                 }
524         }
525
526         end_critical_section(S_NETCONFIGS);
527 }
528
529 eNextState RSSAggregator_AnalyseReply(AsyncIO *IO)
530 {
531         HashPos *it = NULL;
532         long len;
533         const char *Key;
534         pRSSConfig *pCfg;
535         u_char rawdigest[MD5_DIGEST_LEN];
536         struct MD5Context md5context;
537         StrBuf *guid;
538         rss_aggregator *Ctx = (rss_aggregator *) IO->Data;
539
540
541         if ((IO->HttpReq.httpcode >= 300) &&
542             (IO->HttpReq.httpcode < 400)  && 
543             (Ctx->RedirectUrl != NULL)) {
544
545                 StrBuf *ErrMsg;
546                 long lens[2];
547                 const char *strs[2];
548
549                 SetRSSState(IO, eRSSFailure);
550                 ErrMsg = NewStrBuf();
551                 if (IO) EVRSSC_syslog(LOG_ALERT, "need a 200, got a %ld !\n",
552                               IO->HttpReq.httpcode);
553                 strs[0] = ChrPtr(Ctx->Url);
554                 lens[0] = StrLength(Ctx->Url);
555
556                 strs[1] = ChrPtr(Ctx->rooms);
557                 lens[1] = StrLength(Ctx->rooms);
558
559                 if (IO->HttpReq.CurlError == NULL)
560                         IO->HttpReq.CurlError = "";
561
562                 StrBufPrintf(ErrMsg,
563                              "Error while RSS-Aggregation Run of %s\n"
564                              " need a 200, got a %ld !\n"
565                              " Curl Error message: \n%s / %s\n"
566                              " Redirect header points to: %s\n"
567                              " Response text was: \n"
568                              " \n %s\n",
569                              ChrPtr(Ctx->Url),
570                              IO->HttpReq.httpcode,
571                              IO->HttpReq.errdesc,
572                              IO->HttpReq.CurlError,
573                              ChrPtr(Ctx->RedirectUrl),
574                              ChrPtr(IO->HttpReq.ReplyData)
575                         );
576
577                 CtdlAideFPMessage(
578                         ChrPtr(ErrMsg),
579                         "RSS Aggregation run failure",
580                         2, strs, (long*) &lens,
581                         CCID,
582                         IO->ID,
583                         EvGetNow(IO));
584                 
585                 FreeStrBuf(&ErrMsg);
586                 EVRSSC_syslog(LOG_DEBUG,
587                               "RSS feed returned an invalid http status code. <%s><HTTP %ld>\n",
588                               ChrPtr(Ctx->Url),
589                               IO->HttpReq.httpcode);
590                 return eAbort;
591         }
592         else if (IO->HttpReq.httpcode != 200)
593         {
594                 StrBuf *ErrMsg;
595                 long lens[2];
596                 const char *strs[2];
597
598                 SetRSSState(IO, eRSSFailure);
599                 ErrMsg = NewStrBuf();
600                 if (IO) EVRSSC_syslog(LOG_ALERT, "need a 200, got a %ld !\n",
601                               IO->HttpReq.httpcode);
602                 strs[0] = ChrPtr(Ctx->Url);
603                 lens[0] = StrLength(Ctx->Url);
604
605                 strs[1] = ChrPtr(Ctx->rooms);
606                 lens[1] = StrLength(Ctx->rooms);
607
608                 if (IO->HttpReq.CurlError == NULL)
609                         IO->HttpReq.CurlError = "";
610
611                 StrBufPrintf(ErrMsg,
612                              "Error while RSS-Aggregation Run of %s\n"
613                              " need a 200, got a %ld !\n"
614                              " Curl Error message: \n%s / %s\n"
615                              " Response text was: \n"
616                              " \n %s\n",
617                              ChrPtr(Ctx->Url),
618                              IO->HttpReq.httpcode,
619                              IO->HttpReq.errdesc,
620                              IO->HttpReq.CurlError,
621                              ChrPtr(IO->HttpReq.ReplyData)
622                         );
623
624                 CtdlAideFPMessage(
625                         ChrPtr(ErrMsg),
626                         "RSS Aggregation run failure",
627                         2, strs, (long*) &lens,
628                         CCID,
629                         IO->ID,
630                         EvGetNow(IO));
631                 
632                 FreeStrBuf(&ErrMsg);
633                 EVRSSC_syslog(LOG_DEBUG,
634                               "RSS feed returned an invalid http status code. <%s><HTTP %ld>\n",
635                               ChrPtr(Ctx->Url),
636                               IO->HttpReq.httpcode);
637                 return eAbort;
638         }
639
640         pCfg = &Ctx->Cfg;
641
642         while (pCfg != NULL)
643         {
644                 UpdateLastKnownGood (pCfg, EvGetNow(IO));
645                 if ((Ctx->roomlist_parts > 1) && 
646                     (it == NULL))
647                 {
648                         it = GetNewHashPos(RSSFetchUrls, 0);
649                 }
650                 if (it != NULL)
651                 {
652                         void *vptr;
653                         if (GetNextHashPos(Ctx->OtherQRnumbers, it, &len, &Key, &vptr))
654                                 pCfg = vptr;
655                         else
656                                 pCfg = NULL;
657                 }
658                 else 
659                         pCfg = NULL;
660         }
661         DeleteHashPos (&it);
662
663         SetRSSState(IO, eRSSUT);
664
665         MD5Init(&md5context);
666
667         MD5Update(&md5context,
668                   (const unsigned char*)SKEY(IO->HttpReq.ReplyData));
669
670         MD5Update(&md5context,
671                   (const unsigned char*)SKEY(Ctx->Url));
672
673         MD5Final(rawdigest, &md5context);
674         guid = NewStrBufPlain(NULL,
675                               MD5_DIGEST_LEN * 2 + 12 /* _rss2ctdl*/);
676         StrBufHexEscAppend(guid, NULL, rawdigest, MD5_DIGEST_LEN);
677         StrBufAppendBufPlain(guid, HKEY("_rssFM"), 0);
678         if (StrLength(guid) > 40)
679                 StrBufCutAt(guid, 40, NULL);
680         /* Find out if we've already seen this item */
681
682 #ifndef DEBUG_RSS
683
684         if (CheckIfAlreadySeen("RSS Whole",
685                                guid,
686                                EvGetNow(IO),
687                                EvGetNow(IO) - USETABLE_ANTIEXPIRE,
688                                eUpdate,
689                                CCID, IO->ID)
690             != 0)
691         {
692                 FreeStrBuf(&guid);
693
694                 EVRSSC_syslog(LOG_DEBUG, "RSS feed already seen. <%s>\n", ChrPtr(Ctx->Url));
695                 return eAbort;
696         }
697         FreeStrBuf(&guid);
698 #endif
699         SetRSSState(IO, eRSSParsing);
700         return RSSAggregator_ParseReply(IO);
701 }
702
703 eNextState RSSAggregator_FinishHttp(AsyncIO *IO)
704 {
705         return CurlQueueDBOperation(IO, RSSAggregator_AnalyseReply);
706 }
707
708 /*
709  * Begin a feed parse
710  */
711 int rss_do_fetching(rss_aggregator *RSSAggr)
712 {
713         AsyncIO         *IO = &RSSAggr->IO;
714         rss_item *ri;
715         time_t now;
716         CURLcode sta;
717         CURL *chnd;
718
719
720         now = time(NULL);
721
722         if ((RSSAggr->next_poll != 0) && (now < RSSAggr->next_poll))
723                 return 0;
724
725         ri = (rss_item*) malloc(sizeof(rss_item));
726         memset(ri, 0, sizeof(rss_item));
727         RSSAggr->Item = ri;
728
729         if (! InitcURLIOStruct(&RSSAggr->IO,
730                                RSSAggr,
731                                "Citadel RSS Client",
732                                RSSAggregator_FinishHttp,
733                                RSSAggregator_Terminate,
734                                RSSAggregator_TerminateDB,
735                                RSSAggregator_ShutdownAbort))
736         {
737                 EVRSSCM_syslog(LOG_ALERT, "Unable to initialize libcurl.\n");
738                 return 0;
739         }
740         chnd = IO->HttpReq.chnd;
741         OPT(HEADERDATA, IO);
742         OPT(HEADERFUNCTION, GetLocationString);
743         SetRSSState(IO, eRSSCreated);
744
745         safestrncpy(((CitContext*)RSSAggr->IO.CitContext)->cs_host,
746                     ChrPtr(RSSAggr->Url),
747                     sizeof(((CitContext*)RSSAggr->IO.CitContext)->cs_host));
748
749         EVRSSC_syslog(LOG_DEBUG, "Fetching RSS feed <%s>\n", ChrPtr(RSSAggr->Url));
750         ParseURL(&RSSAggr->IO.ConnectMe, RSSAggr->Url, 80);
751         CurlPrepareURL(RSSAggr->IO.ConnectMe);
752
753         SetRSSState(IO, eRSSFetching);
754         QueueCurlContext(&RSSAggr->IO);
755         return 1;
756 }
757
758 /*
759  * Scan a room's netconfig to determine whether it is requesting any RSS feeds
760  */
761 void rssclient_scan_room(struct ctdlroom *qrbuf, void *data, OneRoomNetCfg *OneRNCFG)
762 {
763         const RSSCfgLine *RSSCfg = (RSSCfgLine *)OneRNCFG->NetConfigs[rssclient];
764         rss_aggregator *RSSAggr = NULL;
765         rss_aggregator *use_this_RSSAggr = NULL;
766         void *vptr;
767
768         pthread_mutex_lock(&RSSQueueMutex);
769         if (GetHash(RSSQueueRooms, LKEY(qrbuf->QRnumber), &vptr))
770         {
771                 EVRSSQ_syslog(LOG_DEBUG,
772                               "rssclient: [%ld] %s already in progress.\n",
773                               qrbuf->QRnumber,
774                               qrbuf->QRname);
775                 pthread_mutex_unlock(&RSSQueueMutex);
776                 return;
777         }
778         pthread_mutex_unlock(&RSSQueueMutex);
779
780         if (server_shutting_down) return;
781
782         while (RSSCfg != NULL)
783         {
784                 pthread_mutex_lock(&RSSQueueMutex);
785                 GetHash(RSSFetchUrls,
786                         SKEY(RSSCfg->Url),
787                         &vptr);
788
789                 use_this_RSSAggr = (rss_aggregator *)vptr;
790                 if (use_this_RSSAggr != NULL)
791                 {
792                         pRSSConfig *pRSSCfg;
793
794                         StrBufAppendBufPlain(
795                                 use_this_RSSAggr->rooms,
796                                 qrbuf->QRname,
797                                 -1, 0);
798                         if (use_this_RSSAggr->roomlist_parts==1)
799                         {
800                                 use_this_RSSAggr->OtherQRnumbers
801                                         = NewHash(1, lFlathash);
802                         }
803
804                         pRSSCfg = (pRSSConfig *) malloc(sizeof(pRSSConfig));
805
806                         pRSSCfg->QRnumber = qrbuf->QRnumber;
807                         pRSSCfg->pCfg = RSSCfg;
808
809                         Put(use_this_RSSAggr->OtherQRnumbers,
810                             LKEY(qrbuf->QRnumber),
811                             pRSSCfg,
812                             NULL);
813                         use_this_RSSAggr->roomlist_parts++;
814
815                         pthread_mutex_unlock(&RSSQueueMutex);
816
817                         RSSCfg = RSSCfg->next;
818                         continue;
819                 }
820                 pthread_mutex_unlock(&RSSQueueMutex);
821
822                 RSSAggr = (rss_aggregator *) malloc(
823                         sizeof(rss_aggregator));
824
825                 memset (RSSAggr, 0, sizeof(rss_aggregator));
826                 RSSAggr->Cfg.QRnumber = qrbuf->QRnumber;
827                 RSSAggr->Cfg.pCfg = RSSCfg;
828                 RSSAggr->roomlist_parts = 1;
829                 RSSAggr->Url = NewStrBufDup(RSSCfg->Url);
830
831                 RSSAggr->ItemType = RSS_UNSET;
832
833                 RSSAggr->rooms = NewStrBufPlain(
834                         qrbuf->QRname, -1);
835
836                 pthread_mutex_lock(&RSSQueueMutex);
837
838                 Put(RSSFetchUrls,
839                     SKEY(RSSAggr->Url),
840                     RSSAggr,
841                     DeleteRssCfg);
842
843                 pthread_mutex_unlock(&RSSQueueMutex);
844                 RSSCfg = RSSCfg->next;
845         }
846 }
847
848 /*
849  * Scan for rooms that have RSS client requests configured
850  */
851 void rssclient_scan(void) {
852         int RSSRoomCount, RSSCount;
853         rss_aggregator *rptr = NULL;
854         void *vrptr = NULL;
855         HashPos *it;
856         long len;
857         const char *Key;
858         time_t now = time(NULL);
859
860         /* Run no more than once every 15 minutes. */
861         if ((now - last_run) < 900) {
862                 EVRSSQ_syslog(LOG_DEBUG,
863                               "Client: polling interval not yet reached; last run was %ldm%lds ago",
864                               ((now - last_run) / 60),
865                               ((now - last_run) % 60)
866                 );
867                 return;
868         }
869
870         /*
871          * This is a simple concurrency check to make sure only one rssclient
872          * run is done at a time.
873          */
874         pthread_mutex_lock(&RSSQueueMutex);
875         RSSCount = GetCount(RSSFetchUrls);
876         RSSRoomCount = GetCount(RSSQueueRooms);
877         pthread_mutex_unlock(&RSSQueueMutex);
878
879         if ((RSSRoomCount > 0) || (RSSCount > 0)) {
880                 EVRSSQ_syslog(LOG_DEBUG,
881                               "rssclient: concurrency check failed; %d rooms and %d url's are queued",
882                               RSSRoomCount, RSSCount
883                         );
884                 return;
885         }
886
887         become_session(&rss_CC);
888         EVRSSQM_syslog(LOG_DEBUG, "rssclient started");
889         CtdlForEachNetCfgRoom(rssclient_scan_room, NULL, rssclient);
890
891         if (GetCount(RSSFetchUrls) > 0)
892         {
893                 pthread_mutex_lock(&RSSQueueMutex);
894                 EVRSSQ_syslog(LOG_DEBUG,
895                                "rssclient starting %d Clients",
896                                GetCount(RSSFetchUrls));
897                 
898                 it = GetNewHashPos(RSSFetchUrls, 0);
899                 while (!server_shutting_down &&
900                        GetNextHashPos(RSSFetchUrls, it, &len, &Key, &vrptr) &&
901                        (vrptr != NULL)) {
902                         rptr = (rss_aggregator *)vrptr;
903                         if (!rss_do_fetching(rptr))
904                                 UnlinkRSSAggregator(rptr);
905                 }
906                 DeleteHashPos(&it);
907                 pthread_mutex_unlock(&RSSQueueMutex);
908         }
909         else
910                 EVRSSQM_syslog(LOG_DEBUG, "Nothing to do.");
911
912         EVRSSQM_syslog(LOG_DEBUG, "rssclient ended\n");
913         return;
914 }
915
916 void rss_cleanup(void)
917 {
918         /* citthread_mutex_destroy(&RSSQueueMutex); TODO */
919         DeleteHash(&RSSFetchUrls);
920         DeleteHash(&RSSQueueRooms);
921 }
922
923 void LogDebugEnableRSSClient(const int n)
924 {
925         RSSClientDebugEnabled = n;
926 }
927
928
929 typedef struct __RSSVetoInfo {
930         StrBuf *ErrMsg;
931         time_t Now;
932         int Veto;
933 }RSSVetoInfo;
934
935 void rssclient_veto_scan_room(struct ctdlroom *qrbuf, void *data, OneRoomNetCfg *OneRNCFG)
936 {
937         RSSVetoInfo *Info = (RSSVetoInfo *) data;
938         const RSSCfgLine *RSSCfg = (RSSCfgLine *)OneRNCFG->NetConfigs[rssclient];
939
940         while (RSSCfg != NULL)
941         {
942                 if ((RSSCfg->last_known_good != 0) &&
943                     (RSSCfg->last_known_good + USETABLE_ANTIEXPIRE < Info->Now))
944                 {
945                         StrBufAppendPrintf(Info->ErrMsg,
946                                            "RSS feed not seen for a %d days:: <",
947                                            (Info->Now - RSSCfg->last_known_good) / (24 * 60 * 60));
948
949                         StrBufAppendBuf(Info->ErrMsg, RSSCfg->Url, 0);
950                         StrBufAppendBufPlain(Info->ErrMsg, HKEY(">\n"), 0);
951                 }
952                 RSSCfg = RSSCfg->next;
953         }
954 }
955
956 int RSSCheckUsetableVeto(StrBuf *ErrMsg)
957 {
958         RSSVetoInfo Info;
959
960         Info.ErrMsg = ErrMsg;
961         Info.Now = time (NULL);
962         Info.Veto = 0;
963
964         CtdlForEachNetCfgRoom(rssclient_veto_scan_room, &Info, rssclient);
965
966         return Info.Veto;;
967 }
968
969
970
971
972 void ParseRSSClientCfgLine(const CfgLineType *ThisOne, StrBuf *Line, const char *LinePos, OneRoomNetCfg *OneRNCFG)
973 {
974         RSSCfgLine *RSSCfg;
975
976         RSSCfg = (RSSCfgLine *) malloc (sizeof(RSSCfgLine));
977         RSSCfg->Url = NewStrBufPlain (NULL, StrLength (Line));
978         
979
980         StrBufExtract_NextToken(RSSCfg->Url, Line, &LinePos, '|');
981         RSSCfg->last_known_good = StrBufExtractNext_long(Line, &LinePos, '|');
982
983
984         RSSCfg->next = (RSSCfgLine *)OneRNCFG->NetConfigs[ThisOne->C];
985         OneRNCFG->NetConfigs[ThisOne->C] = (RoomNetCfgLine*) RSSCfg;
986 }
987
988 void SerializeRSSClientCfgLine(const CfgLineType *ThisOne, StrBuf *OutputBuffer, OneRoomNetCfg *RNCfg, RoomNetCfgLine *data)
989 {
990         RSSCfgLine *RSSCfg = (RSSCfgLine*) data;
991
992         StrBufAppendBufPlain(OutputBuffer, CKEY(ThisOne->Str), 0);
993         StrBufAppendBufPlain(OutputBuffer, HKEY("|"), 0);
994         StrBufAppendBufPlain(OutputBuffer, SKEY(RSSCfg->Url), 0);
995         StrBufAppendPrintf(OutputBuffer, "|%ld\n", RSSCfg->last_known_good);
996 }
997
998 void DeleteRSSClientCfgLine(const CfgLineType *ThisOne, RoomNetCfgLine **data)
999 {
1000         RSSCfgLine *RSSCfg = (RSSCfgLine*) *data;
1001
1002         FreeStrBuf(&RSSCfg->Url);
1003         free(*data);
1004         *data = NULL;
1005 }
1006
1007
1008 CTDL_MODULE_INIT(rssclient)
1009 {
1010         if (!threading)
1011         {
1012                 CtdlRegisterTDAPVetoHook (RSSCheckUsetableVeto, CDB_USETABLE, 0);
1013
1014                 CtdlREGISTERRoomCfgType(rssclient, ParseRSSClientCfgLine, 0, 1, SerializeRSSClientCfgLine, DeleteRSSClientCfgLine);
1015                 pthread_mutex_init(&RSSQueueMutex, NULL);
1016                 RSSQueueRooms = NewHash(1, lFlathash);
1017                 RSSFetchUrls = NewHash(1, NULL);
1018                 syslog(LOG_INFO, "%s\n", curl_version());
1019                 CtdlRegisterSessionHook(rssclient_scan, EVT_TIMER, PRIO_AGGR + 300);
1020                 CtdlRegisterEVCleanupHook(rss_cleanup);
1021                 CtdlRegisterDebugFlagHook(HKEY("rssclient"), LogDebugEnableRSSClient, &RSSClientDebugEnabled);
1022         }
1023         else
1024         {
1025                 CtdlFillSystemContext(&rss_CC, "rssclient");
1026         }
1027         return "rssclient";
1028 }