/*
 * DSPOOLOUT.C
 *
 * Flush the outbound queue files and start up dnewslinks as appropriate
 *
 * (c)Copyright 1997, Matthew Dillon, All Rights Reserved.  Refer to
 *    the COPYRIGHT file in the base directory of this distribution 
 *    for specific rights granted.
 */

#include "defs.h"

#define TYPE_NORMAL	1
#define TYPE_XREPLIC	2

#define SF_NOSTREAM	0x0001
#define SF_REALTIME	0x0002
#define SF_NOBATCH	0x0004
#define SF_HEADFEED	0x0008
#define SF_NOCHECK	0x0010

int MaxRun = 2;		/* default MaxRun				       */
int MinFlushSecs = 0;	/* minimum time between flushes if queue not caught up */

typedef struct DNode {
    struct DNode *no_Next;
    char	*no_SpoolFile;
} DNode;

DNode *Base;

int spool_file(char *spoolFile, char *hostName, int maxq, int maxr, int qskip, int port, int delay, int type, int flags, char *obip, int txbufsiz, int rxbufsiz, int pass);
DNode *FindNode(char *name);
void AddNode(char *name);

int VerboseOpt;
int ForReal = 1;
int Quiet;
int  TxBufSize = 0;
int  RxBufSize = 0;
char *OutboundIpStr = "-nop";

int
main(int ac, char **av)
{
    char buf[256];
    FILE *fi;
    int pass;
    int sfCount = 200;
    int *sfAry = malloc(sfCount * sizeof(int));
    char *qFile = NULL;
    char *cFile = NNTPSPOOLCTL;

    LoadDiabloConfig();

    /*
     * Shift into the out.going directory
     */

    if (chdir(OutGoing) != 0) {
	fprintf(stderr, "unable to chdir to %s\n", OutGoing);
	exit(1);
    }

    {
	int i;

	for (i = 1; i < ac; ++i) {
	    char *ptr = av[i];
	    if (*ptr == '-') {
		ptr += 2;
		switch(ptr[-1]) {
		case 'q':
		    Quiet = 1;
		    break;
		case 'v':
		    VerboseOpt = 1;
		    break;
		case 'n':
		    ForReal = 0;
		    break;
		case 'f':
		    cFile = (*ptr) ? ptr : av[++i];
		    break;
		case 'T':
		    {
			TxBufSize = strtol((*ptr ? ptr : av[++i]), NULL, 0);
		    }
		    break;
		case 'R':
		    {
			RxBufSize = strtol((*ptr ? ptr : av[++i]), NULL, 0);
		    }
		    break;
		case 'B':
		    {
			if (*ptr == 0)
			    ptr = av[++i];
			OutboundIpStr = malloc(strlen(ptr) + 8);
			sprintf(OutboundIpStr, "-B%s", ptr);
		    }
		    break;
		case 's':
		    MinFlushSecs = strtol((*ptr ? ptr : av[++i]), NULL, 0) * 60;
		    break;
		case 'm':
		    MaxRun = strtol((*ptr ? ptr : av[++i]), NULL, 0);
		    break;
		default:
		    break;
		}
	    } else {
		qFile = ptr;
	    }
	}
    }

    if (cFile[0] != '/') {
	char *p = malloc(strlen(cFile) + strlen(NewsCfg) + 4);
	sprintf(p, "%s/%s", NewsCfg, cFile);
	cFile = p;
    }

    /*
     * Get list of spoolfile names and NNTP hosts, one pair per line,
     * blank lines and lines beginning with '#' excepted.
     */

    for (pass = 1; pass <= 2; ++pass) {
	if ((fi = fopen(cFile, "r")) != NULL) {
	    int count = 0;

	    while (fgets(buf, sizeof(buf), fi) != NULL) {
		char *spoolFile;
		char *hostName;

		if (buf[0] == '\n' || buf[0] == '#')
		    continue;
		if ((spoolFile = strtok(buf, " \t\n")) != NULL &&
		    (hostName = strtok(NULL, " \t\n")) != NULL &&
		    (qFile == NULL || strcmp(spoolFile, qFile) == 0)
		) {
		    int maxq = -1;
		    int port = -1;
		    int maxr = MaxRun;
		    int type = TYPE_NORMAL;
		    int delay = 0;
		    int flags = 0;
		    int qskip = 0;
		    int txbufsiz = TxBufSize;
		    int rxbufsiz = RxBufSize;
		    char *obip = NULL;

		    {
			char *maxqStr;

			while ((maxqStr = strtok(NULL, " \t\n")) != NULL) {
			    if (strtol(maxqStr, NULL, 0) > 0) {
				maxq = strtol(maxqStr, NULL, 0);
			    } else if (strcmp(maxqStr, "xreplic") == 0) {
				type = TYPE_XREPLIC;
			    } else if (strcmp(maxqStr, "nostream") == 0) {
				flags |= SF_NOSTREAM;
			    } else if (strcmp(maxqStr, "headfeed") == 0) {
				flags |= SF_HEADFEED;
			    } else if (strcmp(maxqStr, "nocheck") == 0) {
				flags |= SF_NOCHECK;
			    } else if (strcmp(maxqStr, "realtime") == 0) {
				flags |= SF_REALTIME;
			    } else if (strcmp(maxqStr, "nobatch") == 0) {
				flags |= SF_NOBATCH;
			    } else if (strncmp(maxqStr, "bind=", 5) == 0) {
				obip = maxqStr + 5;
			    } else if (maxqStr[0] == 'q') {
				qskip = strtol(maxqStr + 1, NULL, 0);
			    } else if (maxqStr[0] == 'n') {
				maxr = strtol(maxqStr + 1, NULL, 0);
			    } else if (maxqStr[0] == 'p') {
				port = strtol(maxqStr + 1, NULL, 0);
			    } else if (maxqStr[0] == 'd') {
				delay = strtol(maxqStr + 1, NULL, 0);
			    } else if (maxqStr[0] == 'T') {
				txbufsiz = strtol(maxqStr + 1, NULL, 0);
			    } else if (maxqStr[0] == 'R') {
				rxbufsiz = strtol(maxqStr + 1, NULL, 0);
			    } else {
				if (pass == 1)
				    fprintf(stderr, "bad keyword in control file: %s\n", maxqStr);
			    }
			}
		    }

		    /*
		     * run spool file, with flags modifier from pass 1.  Pass1
		     * may clear the SF_REALTIME flag.
		     */
		    if (count >= sfCount) {
			sfCount += 200;
			sfAry = realloc(sfAry, sfCount * sizeof(int));
			memset(sfAry+count, 0, (sfCount - count) * sizeof(int));
		    }
		    if (pass == 2) {
			flags = (flags & ~SF_REALTIME) | 
				(sfAry[count] & SF_REALTIME);
		    }

		    flags = spool_file(
			spoolFile, 
			hostName,
			maxq,
			maxr,
			qskip,
			port,
			delay,
			type, 
			flags,
			obip,
			txbufsiz,
			rxbufsiz,
			pass
		    );

		    if (pass == 1)
			sfAry[count] = flags;
		    ++count;
		}
	    }
	    fclose(fi);
	} else if (pass == 1) {
	    fprintf(stderr, "Unable to open %s/" NNTPSPOOLCTL "\n", NewsCfg);
	}
	if (pass == 1) {
	    char sysline[256];

	    sprintf(sysline, "%s/dicmd flush", NEWSBIN);
	    system(sysline);
	}
    }
    return(0);
}

/*
 * SPOOL_FILE() - pass 1 - rename spool files
 *
 *		  pass 2 - start dnewslink processes
 */

int
spool_file(char *spoolFile, char *hostName, int maxq, int maxr, int qskip, int port, int delay, int type, int flags, char *obip, int txbufsiz, int rxbufsiz, int pass)
{
    char seqFile[256];
    char newFile[256];
    char obIpBuf[256];
    char *outboundIpStr = OutboundIpStr;
    int begSeq = 0;
    int endSeq = 0;
    long newTime = 0;
    char portBuf[32];
    int fd;
    struct stat st;
    char txBufSizeStr[32];
    char rxBufSizeStr[32];

    /*
     * Initialize rx/txBufSizeStr
     */

    sprintf(txBufSizeStr, "-nop");
    sprintf(rxBufSizeStr, "-nop");

    if (txbufsiz > 0 && txbufsiz < 16 * 1024 * 1024)
	sprintf(txBufSizeStr, "-T%d", txbufsiz);
    if (rxbufsiz > 0 && rxbufsiz < 16 * 1024 * 1024)
	sprintf(rxBufSizeStr, "-R%d", rxbufsiz);

    /*
     * Override outbound IP
     */

    if (obip != NULL && strlen(obip) < sizeof(obIpBuf) - 8) {
	sprintf(obIpBuf, "-B%s", obip);
	outboundIpStr = obIpBuf;
    }

    if (port < 0)
	port = 119;
    sprintf(portBuf, "%d", port);

    sprintf(seqFile, ".%s.seq", spoolFile);

    /*
     * Get beginning and ending sequence numbers
     */

    fd = open(seqFile, O_RDWR|O_CREAT, 0644);

    if (fd >= 0) {
	char buf[64];

	xflock(fd, XLOCK_EX);		/* lock and leave locked */
	memset(buf, 0, sizeof(buf));
	read(fd, buf, sizeof(buf) - 1);
	sscanf(buf, "%d %d %lx", &begSeq, &endSeq, &newTime);
    }

    /*
     * Discard beginning sequence numbers that are now deleted, or
     * delete queue files that are too old.
     */

    while (begSeq < endSeq) {
	sprintf(newFile, "%s.S%05d", spoolFile, begSeq);
	if (maxq > 0 && begSeq < endSeq - maxq)
	    remove(newFile);
	if (stat(newFile, &st) == 0)
	    break;
	++begSeq;
    }

    /*
     * If primary spool file exists, shift to primary
     * queue
     */

    if (pass == 1) {
	{
	    int tries = 100;
	    int32 dt;

	    while (tries > 0) {
		struct stat st;

		sprintf(newFile, "%s.S%05d", spoolFile, endSeq);
		if (stat(newFile, &st) < 0)
		    break;
		--tries;
		++endSeq;
	    }

	    /*
	     * Only create a new spool file if:
	     *
	     * (1) endSeq == begSeq + 1, or
	     * (2) current time larger then newTime + MinFlushSecs
	     * (3) there is time weirdness
	     */

	    dt = time(NULL) - ((int32)newTime + MinFlushSecs);

	    if (endSeq <= begSeq + 4 || dt > 0 || dt < -MinFlushSecs) {
		/*
		 * If a dnewslink has a lock on the unsequenced spool file,
		 * clear the SF_REALTIME flag... a realtime dnewslink is
		 * already running so there is no sense trying to start 
		 * another one.
		 */
		if (flags & SF_REALTIME) {
		    int tfd;

		    if ((tfd = open(spoolFile, O_RDWR)) >= 0) {
			if (xflock(tfd, XLOCK_EX|XLOCK_NB) < 0) {
			    if (VerboseOpt)
				printf("%s: realtime already\n", spoolFile);
			    flags &= ~SF_REALTIME;
			}
			close(tfd);
		    }
		}

		if (rename(spoolFile, newFile) == 0) {
		    ++endSeq;
		    if (!Quiet)
			printf("dspoolout: add %s\n", newFile);
		} else {
		    if (VerboseOpt && !Quiet)
			printf("dspoolout: nofile %s\n", newFile);
		    if (endSeq == begSeq)
			AddNode(spoolFile);
		}
		newTime = time(NULL);
	    } else {
		AddNode(spoolFile);
		if (VerboseOpt && !Quiet)
		    printf("dspoolout: wait %s\n", spoolFile);
	    }
	}

	/*
	 * Update sequence number file
	 */

	if (fd >= 0) {
	    char buf[64];

	    sprintf(buf, "%5d %5d %08lx\n", begSeq, endSeq, newTime);
	    lseek(fd, 0L, 0);
	    write(fd, buf, strlen(buf));
	    ftruncate(fd, lseek(fd, 0L, 1));
	}
    }

    if (fd >= 0) {
	xflock(fd, XLOCK_UN);
	close(fd);
    }

    /*
     * Run up to N newslink's/innxmit's, usually set at 2.  More may actually
     * be running: the nntplink from the INN channel, and older nntplink's
     * from previously flushed INN channels that have not reached EOF
     * on their pipe.
     *
     * Note that we spool out the queue as a FIFO.  This is important as
     * it reduces article jumble.
     */

    if (pass == 2 && FindNode(spoolFile) == NULL) {
	int tries = maxr;
	int look;

	/*
	 * Start up to maxr dnewslinks.  If the nobatch
	 * flag is set, do not start any.
	 */

	if (flags & SF_NOBATCH)
	    tries = 0;

	for (look = 0; look < 10 && tries && begSeq + look < endSeq - qskip; ++look) {
	    int use = begSeq + look;
	    int fd;
	    char numBuf[32];

	    if (qskip)
		sprintf(numBuf, "%d", endSeq - use - qskip);
	    else
		sprintf(numBuf, "32");

	    sprintf(newFile, "%s.S%05d", spoolFile, use);
	    if ((fd = open(newFile, O_RDWR)) >= 0) {
		if (xflock(fd, XLOCK_EX|XLOCK_NB) == 0) {
		    char seqName[64];
		    char templateFile[256];

		    xflock(fd, XLOCK_UN);

		    if (!Quiet)
			printf("dspoolout: run %s.S%05d\n", spoolFile, use);

		    if (ForReal && fork() == 0) {
			if (delay > 0)
			    sleep(delay);

			sprintf(templateFile, "%s.S%%05d", spoolFile);
			sprintf(seqName, "%d", use);

			switch(type) {
			case TYPE_NORMAL:
			    execl(NEWSBIN "/dnewslink", "dnewslink", 
				"-h", hostName,
				"-b", templateFile,
				"-S", seqName,
				"-N", numBuf,
				"-P", portBuf,
				"-D",
				((flags & SF_NOSTREAM) ? "-i" : "-nop"),
				((flags & SF_HEADFEED) ? "-H" : "-nop"),
				((flags & SF_NOCHECK) ? "-I" : "-nop"),
				outboundIpStr,
				txBufSizeStr,
				rxBufSizeStr,
				NULL
			    );
			    break;
			case TYPE_XREPLIC:
#ifdef NOTDEF
		/* not supported with new format */
			    execlp("innxmit", "innxmit",
				"-a",
				"-r",
				"-S", hostName,
				newFile,
				NULL
			    );
#endif
			    break;
			}
			exit(0);
		    }
		}
		--tries;	/* even if lock fails */
		close(fd);
	    }
	}

	/*
	 * Deal with the realtime dnewslink, but only if:
	 *	The realtime flag is set
	 *	We have not gotten behind
	 *	We did not detect another realtime process
	 */

	if (VerboseOpt)
	    printf("%s: flags %02x nseq %d\n", spoolFile, flags & SF_REALTIME, endSeq-begSeq);

	if ((flags & SF_REALTIME) && endSeq - begSeq <= 2) {
	    int fd;

	    if ((fd = open(spoolFile, O_RDWR|O_CREAT, 0600)) >= 0) {
		if (xflock(fd, XLOCK_EX|XLOCK_NB) == 0) {
		    xflock(fd, XLOCK_UN);

		    if (!Quiet)
			printf("dspoolout: run realtime %s\n", spoolFile);

		    if (ForReal && fork() == 0) {
			if (delay > 0)
			    sleep(delay);

			switch(type) {
			case TYPE_NORMAL:
			    /*
			     * realtime option is '-r'.  If a delay of -1
			     * is specified, we pass -r-1 to dnewslink which
			     * reduces the tail latency from 1 second to 1ms.
			     */
			    execl(NEWSBIN "/dnewslink", "dnewslink", 
				"-h", hostName,
				"-b", spoolFile,
				"-P", portBuf,
				"-N", "100",	/* 1000 spool cycles	*/
				((delay < 0) ? "-r-1" : "-r"),
				((flags & SF_NOSTREAM) ? "-i" : "-nop"),
				((flags & SF_HEADFEED) ? "-H" : "-nop"),
				((flags & SF_NOCHECK) ? "-I" : "-nop"),
				outboundIpStr,
				txBufSizeStr,
				rxBufSizeStr,
				NULL
			    );
			    break;
			case TYPE_XREPLIC:
			    break;
			}
			exit(0);
		    }
		}
		close(fd);
	    }
	}
    }
    return(flags);
}

DNode *
FindNode(char *name)
{
    DNode *node;

    for (node = Base; node; node = node->no_Next) {
	if (strcmp(name, node->no_SpoolFile) == 0)
	    break;
    }
    return(node);
}

void
AddNode(char *name)
{
    DNode *node = malloc(sizeof(Node) + strlen(name) + 1);

    node->no_Next = Base;
    node->no_SpoolFile = (char *)(node + 1);
    Base = node;
    strcpy(node->no_SpoolFile, name);
}