MessQueue.cpp

/*
 *  XMail by Davide Libenzi (Intranet and Internet mail server)
 *  Copyright (C) 1999,..,2010  Davide Libenzi
 *
 *  This program is free software; you can redistribute it and/or modify
 *  it under the terms of the GNU General Public License as published by
 *  the Free Software Foundation; either version 2 of the License, or
 *  (at your option) any later version.
 *
 *  This program is distributed in the hope that it will be useful,
 *  but WITHOUT ANY WARRANTY; without even the implied warranty of
 *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 *  GNU General Public License for more details.
 *
 *  You should have received a copy of the GNU General Public License
 *  along with this program; if not, write to the Free Software
 *  Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
 *
 *  Davide Libenzi <davidel@xmailserver.org>
 *
 */

#include "SysInclude.h"
#include "SysDep.h"
#include "SvrDefines.h"
#include "ShBlocks.h"
#include "ResLocks.h"
#include "StrUtils.h"
#include "SList.h"
#include "MD5.h"
#include "Base64Enc.h"
#include "BuffSock.h"
#include "MailConfig.h"
#include "UsrUtils.h"
#include "SvrUtils.h"
#include "AppDefines.h"
#include "MiscUtils.h"
#include "SMTPUtils.h"
#include "MessQueue.h"
#include "SMAILUtils.h"

#define QUEF_SHUTDOWN               (1 << 0)

#define QUMF_DELETED                (1 << 0)
#define QUMF_FREEZE                 (1 << 1)

#define QUE_MASK_TMPFLAGS(v)        ((v) & ~(QUMF_DELETED | QUMF_FREEZE))

#define QUE_ARENA_SCAN_INTERVAL     15
#define QUE_ARENA_SCAN_WAIT         2
#define QUE_SCAN_THREAD_MAXWAIT     60

struct MessageQueue {
	SysListHead ReadyQueue;
	SysListHead RsndArenaQueue;
	int iReadyCount;
	int iRsndArenaCount;
	SYS_MUTEX hMutex;
	SYS_EVENT hReadyEvent;
	char *pszRootPath;
	int iMaxRetry;
	int iRetryTimeout;
	int iRetryIncrRatio;
	int iNumDirsLevel;
	unsigned long ulFlags;
	SYS_THREAD hRsndScanThread;
};

struct QueueMessage {
	SysListHead LLink;
	int iLevel1;
	int iLevel2;
	char const *pszQueueDir;
	char *pszFileName;
	int iNumTries;
	time_t tLastTry;
	unsigned long ulFlags;
};

static int QueGetFilePath(MessageQueue *pMQ, QueueMessage *pQM, char *pszFilePath,
			  char const *pszQueueDir = NULL);

static QueueMessage *QueAllocMessage(int iLevel1, int iLevel2, char const *pszQueueDir,
				     char const *pszFileName, int iNumTries, time_t tLastTry)
{
	QueueMessage *pQM = (QueueMessage *) SysAlloc(sizeof(QueueMessage));

	if (pQM == NULL)
		return NULL;

	SYS_INIT_LIST_LINK(&pQM->LLink);
	pQM->iLevel1 = iLevel1;
	pQM->iLevel2 = iLevel2;
	pQM->pszQueueDir = pszQueueDir;
	pQM->pszFileName = SysStrDup(pszFileName);
	pQM->iNumTries = iNumTries;
	pQM->tLastTry = tLastTry;
	pQM->ulFlags = 0;

	return pQM;
}

static int QueFreeMessage(QueueMessage *pQM)
{
	SysFree(pQM->pszFileName);
	SysFree(pQM);

	return 0;
}

static int QueFreeMessList(SysListHead *pHead)
{
	SysListHead *pLLink;

	while ((pLLink = SYS_LIST_FIRST(pHead)) != NULL) {
		QueueMessage *pQM = SYS_LIST_ENTRY(pLLink, QueueMessage, LLink);

		SYS_LIST_DEL(pLLink);
		QueFreeMessage(pQM);
	}

	return 0;
}

static int QueCreateStruct(char const *pszRootPath)
{
	/* Create message dir (new messages queue) */
	char szDirPath[SYS_MAX_PATH];

	StrSNCpy(szDirPath, pszRootPath);
	AppendSlash(szDirPath);
	StrSNCat(szDirPath, QUEUE_MESS_DIR);

	if (!SysExistDir(szDirPath) && SysMakeDir(szDirPath) < 0)
		return ErrGetErrorCode();

	/* Create message resend dir (resend messages queue) */
	StrSNCpy(szDirPath, pszRootPath);
	AppendSlash(szDirPath);
	StrSNCat(szDirPath, QUEUE_RSND_DIR);

	if (!SysExistDir(szDirPath) && SysMakeDir(szDirPath) < 0)
		return ErrGetErrorCode();

	/* Create info dir */
	StrSNCpy(szDirPath, pszRootPath);
	AppendSlash(szDirPath);
	StrSNCat(szDirPath, QUEUE_INFO_DIR);

	if (!SysExistDir(szDirPath) && SysMakeDir(szDirPath) < 0)
		return ErrGetErrorCode();

	/* Create temp dir */
	StrSNCpy(szDirPath, pszRootPath);
	AppendSlash(szDirPath);
	StrSNCat(szDirPath, QUEUE_TEMP_DIR);

	if (!SysExistDir(szDirPath) && SysMakeDir(szDirPath) < 0)
		return ErrGetErrorCode();

	/* Create send log dir */
	StrSNCpy(szDirPath, pszRootPath);
	AppendSlash(szDirPath);
	StrSNCat(szDirPath, QUEUE_SLOG_DIR);

	if (!SysExistDir(szDirPath) && SysMakeDir(szDirPath) < 0)
		return ErrGetErrorCode();

	/* Create custom message processing dir */
	StrSNCpy(szDirPath, pszRootPath);
	AppendSlash(szDirPath);
	StrSNCat(szDirPath, QUEUE_CUST_DIR);

	if (!SysExistDir(szDirPath) && SysMakeDir(szDirPath) < 0)
		return ErrGetErrorCode();

	/* Create user custom message processing dir (mailproc.tab cache) */
	StrSNCpy(szDirPath, pszRootPath);
	AppendSlash(szDirPath);
	StrSNCat(szDirPath, QUEUE_MPRC_DIR);

	if (!SysExistDir(szDirPath) && SysMakeDir(szDirPath) < 0)
		return ErrGetErrorCode();

	/* Create frozen dir */
	StrSNCpy(szDirPath, pszRootPath);
	AppendSlash(szDirPath);
	StrSNCat(szDirPath, QUEUE_FROZ_DIR);

	if (!SysExistDir(szDirPath) && SysMakeDir(szDirPath) < 0)
		return ErrGetErrorCode();

	return 0;
}

static int QueLoadMessageStat(MessageQueue *pMQ, QueueMessage *pQM)
{
	/* Build the slog file path */
	char szSlogFilePath[SYS_MAX_PATH];

	QueGetFilePath(pMQ, pQM, szSlogFilePath, QUEUE_SLOG_DIR);

	/* Try to load message statistics */
	FILE *pLogFile = fopen(szSlogFilePath, "rt");

	if (pLogFile != NULL) {
		int iNumTries = 0;
		unsigned long ulLastTime = 0, ulPeekTime;
		char szLogLine[1024];

		while (MscFGets(szLogLine, sizeof(szLogLine) - 1, pLogFile) != NULL)
			if (sscanf(szLogLine, "[PeekTime] %lu", &ulPeekTime) == 1)
				++iNumTries, ulLastTime = ulPeekTime;

		fclose(pLogFile);

		pQM->iNumTries = iNumTries;
		pQM->tLastTry = (time_t) ulLastTime;
	}

	return 0;
}

static int QueLoadMessages(MessageQueue *pMQ, int iLevel1, int iLevel2)
{
	/* File scan the new messages dir */
	char szDirPath[SYS_MAX_PATH];

	SysSNPrintf(szDirPath, sizeof(szDirPath) - 1, "%s%d%s%d%s%s",
		    pMQ->pszRootPath, iLevel1, SYS_SLASH_STR, iLevel2,
		    SYS_SLASH_STR, QUEUE_MESS_DIR);

	char szMsgFileName[SYS_MAX_PATH];
	FSCAN_HANDLE hFileScan = MscFirstFile(szDirPath, 0, szMsgFileName,
					      sizeof(szMsgFileName));

	if (hFileScan != INVALID_FSCAN_HANDLE) {
		do {
			if (!IsDotFilename(szMsgFileName)) {
				QueueMessage *pQM =
					QueAllocMessage(iLevel1, iLevel2, QUEUE_MESS_DIR,
							szMsgFileName, 0, 0);

				if (pQM != NULL) {
					/* Add the file to the message queue */
					SYS_LIST_ADDT(&pQM->LLink, &pMQ->ReadyQueue);
					++pMQ->iReadyCount;
				}
			}
		} while (MscNextFile(hFileScan, szMsgFileName, sizeof(szMsgFileName)));
		MscCloseFindFile(hFileScan);

		/* Set the mess event if the queue is not empty */
		if (pMQ->iReadyCount > 0)
			SysSetEvent(pMQ->hReadyEvent);
	}
	/* File scan the resend messages dir */
	SysSNPrintf(szDirPath, sizeof(szDirPath) - 1, "%s%d%s%d%s%s",
		    pMQ->pszRootPath, iLevel1, SYS_SLASH_STR, iLevel2,
		    SYS_SLASH_STR, QUEUE_RSND_DIR);

	if ((hFileScan = MscFirstFile(szDirPath, 0, szMsgFileName,
				      sizeof(szMsgFileName))) != INVALID_FSCAN_HANDLE) {
		do {
			if (!IsDotFilename(szMsgFileName)) {
				QueueMessage *pQM = QueAllocMessage(iLevel1, iLevel2,
								    QUEUE_RSND_DIR,
								    szMsgFileName, 0, 0);

				if (pQM != NULL) {
					/* Load message statistics */
					if (QueLoadMessageStat(pMQ, pQM) < 0) {
						SysLogMessage(LOG_LEV_ERROR,
							      "Error loading queue file: '%s%d%s%d%s%s%s%s'\n",
							      pMQ->pszRootPath, iLevel1,
							      SYS_SLASH_STR, iLevel2,
							      SYS_SLASH_STR, QUEUE_RSND_DIR,
							      SYS_SLASH_STR, szMsgFileName);

						QueFreeMessage(pQM);
					} else {
						/* Add the file to the resend queue */
						SYS_LIST_ADDT(&pQM->LLink, &pMQ->RsndArenaQueue);
						++pMQ->iRsndArenaCount;
					}
				}
			}
		} while (MscNextFile(hFileScan, szMsgFileName, sizeof(szMsgFileName)));
		MscCloseFindFile(hFileScan);
	}

	return 0;
}

static int QueLoad(MessageQueue *pMQ)
{
	char szCurrPath[SYS_MAX_PATH];

	for (int i = 0; i < pMQ->iNumDirsLevel; i++) {
		SysSNPrintf(szCurrPath, sizeof(szCurrPath) - 1, "%s%d", pMQ->pszRootPath, i);
		if (!SysExistDir(szCurrPath) && (SysMakeDir(szCurrPath) < 0))
			return ErrGetErrorCode();

		for (int j = 0; j < pMQ->iNumDirsLevel; j++) {
			SysSNPrintf(szCurrPath, sizeof(szCurrPath) - 1, "%s%d%s%d",
				    pMQ->pszRootPath, i, SYS_SLASH_STR, j);
			if (!SysExistDir(szCurrPath) && (SysMakeDir(szCurrPath) < 0))
				return ErrGetErrorCode();

			if (QueCreateStruct(szCurrPath) < 0 ||
			    QueLoadMessages(pMQ, i, j) < 0)
				return ErrGetErrorCode();
		}
	}

	return 0;
}

static time_t QueNextRetryOp(int iNumTries, unsigned int uRetryTimeout,
			     unsigned int uRetryIncrRatio)
{
	unsigned int uNextOp = uRetryTimeout;

	if (uRetryIncrRatio != 0)
		for (int i = 1; i < iNumTries; i++)
			uNextOp += uNextOp / uRetryIncrRatio;

	return (time_t) uNextOp;
}

static bool QueMessageReadyToSend(MessageQueue *pMQ, QueueMessage *pQM)
{
	return (time(NULL) > (pQM->tLastTry +
			      QueNextRetryOp(pQM->iNumTries, (unsigned int) pMQ->iRetryTimeout,
					     (unsigned int) pMQ->iRetryIncrRatio)));
}

static int QueScanRsndArena(MessageQueue *pMQ)
{
	if (SysLockMutex(pMQ->hMutex, SYS_INFINITE_TIMEOUT) < 0)
		return ErrGetErrorCode();

	SysListHead *pLLink;

	SYS_LIST_FOR_EACH(pLLink, &pMQ->RsndArenaQueue) {
		QueueMessage *pQM = SYS_LIST_ENTRY(pLLink, QueueMessage, LLink);

		if (QueMessageReadyToSend(pMQ, pQM)) {
			/* Set the list pointer to the next item */
			pLLink = pLLink->pPrev;

			/* Remove item from resend arena */
			SYS_LIST_DEL(&pQM->LLink);
			--pMQ->iRsndArenaCount;

			/* Add item from resend queue */
			SYS_LIST_ADDT(&pQM->LLink, &pMQ->ReadyQueue);
			++pMQ->iReadyCount;
		}
	}
	if (pMQ->iReadyCount > 0)
		SysSetEvent(pMQ->hReadyEvent);
	SysUnlockMutex(pMQ->hMutex);

	return 0;
}

static unsigned int QueRsndThread(void *pThreadData)
{
	MessageQueue *pMQ = (MessageQueue *) pThreadData;
	int iElapsedTime = 0;

	while ((pMQ->ulFlags & QUEF_SHUTDOWN) == 0) {
		SysSleep(QUE_ARENA_SCAN_WAIT);

		iElapsedTime += QUE_ARENA_SCAN_WAIT;
		if (iElapsedTime > QUE_ARENA_SCAN_INTERVAL) {
			iElapsedTime = 0;

			/* Scan rsnd arena to prepare messages to resend */
			QueScanRsndArena(pMQ);
		}
	}
	pMQ->ulFlags &= ~QUEF_SHUTDOWN;

	return 0;
}

QUEUE_HANDLE QueOpen(char const *pszRootPath, int iMaxRetry, int iRetryTimeout,
		     int iRetryIncrRatio, int iNumDirsLevel)
{
	MessageQueue *pMQ = (MessageQueue *) SysAlloc(sizeof(MessageQueue));

	if (pMQ == NULL)
		return INVALID_QUEUE_HANDLE;

	SYS_INIT_LIST_HEAD(&pMQ->ReadyQueue);
	SYS_INIT_LIST_HEAD(&pMQ->RsndArenaQueue);
	pMQ->iReadyCount = 0;
	pMQ->iRsndArenaCount = 0;
	pMQ->iMaxRetry = iMaxRetry;
	pMQ->iRetryTimeout = iRetryTimeout;
	pMQ->iRetryIncrRatio = iRetryIncrRatio;
	pMQ->iNumDirsLevel = iNumDirsLevel;
	pMQ->ulFlags = 0;

	if ((pMQ->hMutex = SysCreateMutex()) == SYS_INVALID_MUTEX) {
		SysFree(pMQ);
		return INVALID_QUEUE_HANDLE;
	}
	if ((pMQ->hReadyEvent = SysCreateEvent(1)) == SYS_INVALID_EVENT) {
		SysCloseMutex(pMQ->hMutex);
		SysFree(pMQ);
		return INVALID_QUEUE_HANDLE;
	}
	/* Set the queue root path */
	char szRootPath[SYS_MAX_PATH];

	StrSNCpy(szRootPath, pszRootPath);
	AppendSlash(szRootPath);

	pMQ->pszRootPath = SysStrDup(szRootPath);

	/* Load queue status */
	if (QueLoad(pMQ) < 0) {
		ErrorPush();
		SysFree(pMQ->pszRootPath);
		SysCloseEvent(pMQ->hReadyEvent);
		SysCloseMutex(pMQ->hMutex);
		SysFree(pMQ);

		ErrSetErrorCode(ErrorPop());
		return INVALID_QUEUE_HANDLE;
	}
	/* Start rsnd arena scan thread */
	if ((pMQ->hRsndScanThread = SysCreateThread(QueRsndThread, pMQ)) == SYS_INVALID_THREAD) {
		ErrorPush();
		QueFreeMessList(&pMQ->ReadyQueue);
		QueFreeMessList(&pMQ->RsndArenaQueue);
		SysFree(pMQ->pszRootPath);
		SysCloseEvent(pMQ->hReadyEvent);
		SysCloseMutex(pMQ->hMutex);
		SysFree(pMQ);

		ErrSetErrorCode(ErrorPop());
		return INVALID_QUEUE_HANDLE;
	}

	return (QUEUE_HANDLE) pMQ;
}

int QueClose(QUEUE_HANDLE hQueue)
{
	MessageQueue *pMQ = (MessageQueue *) hQueue;

	/* Set the shutdown flag and wait for rsnd scan thread to terminate */
	pMQ->ulFlags |= QUEF_SHUTDOWN;
	SysWaitThread(pMQ->hRsndScanThread, QUE_SCAN_THREAD_MAXWAIT);
	SysCloseThread(pMQ->hRsndScanThread, 1);

	/* Clear queues */
	QueFreeMessList(&pMQ->ReadyQueue);
	QueFreeMessList(&pMQ->RsndArenaQueue);
	SysCloseEvent(pMQ->hReadyEvent);
	SysCloseMutex(pMQ->hMutex);
	SysFree(pMQ->pszRootPath);
	SysFree(pMQ);

	return 0;
}

int QueGetDirsLevel(QUEUE_HANDLE hQueue)
{
	MessageQueue *pMQ = (MessageQueue *) hQueue;

	return pMQ->iNumDirsLevel;
}

char const *QueGetRootPath(QUEUE_HANDLE hQueue)
{
	MessageQueue *pMQ = (MessageQueue *) hQueue;

	return pMQ->pszRootPath;
}

char *QueLoadLastLogEntry(char const *pszLogFilePath)
{
	FILE *pLogFile = fopen(pszLogFilePath, "rb");

	if (pLogFile == NULL) {
		ErrSetErrorCode(ERR_FILE_OPEN, pszLogFilePath);
		return NULL;
	}

	SYS_OFF_T llCurrOffset = 0, llBaseOffset = (SYS_OFF_T) -1, llEndOffset;
	unsigned long ulPeekTime;
	char szLogLine[1024];

	for (;;) {
		llCurrOffset = Sys_ftell(pLogFile);

		if (MscFGets(szLogLine, sizeof(szLogLine) - 1, pLogFile) == NULL)
			break;
		if (sscanf(szLogLine, "[PeekTime] %lu", &ulPeekTime) == 1)
			llBaseOffset = llCurrOffset;
	}
	if (llBaseOffset == (SYS_OFF_T) -1) {
		fclose(pLogFile);
		ErrSetErrorCode(ERR_EMPTY_LOG, pszLogFilePath);
		return NULL;
	}
	/* Get end offset (end of file) */
	Sys_fseek(pLogFile, 0, SEEK_END);
	llEndOffset = Sys_ftell(pLogFile);

	/* Load last entry */
	unsigned int uEntrySize = (unsigned int) (llEndOffset - llBaseOffset);
	char *pszEntry = (char *) SysAlloc(uEntrySize + 1);

	if (pszEntry == NULL) {
		fclose(pLogFile);
		return NULL;
	}
	Sys_fseek(pLogFile, llBaseOffset, SEEK_SET);
	if (!fread(pszEntry, uEntrySize, 1, pLogFile)) {
		SysFree(pszEntry);
		fclose(pLogFile);
		ErrSetErrorCode(ERR_FILE_READ, pszLogFilePath);
		return NULL;
	}
	pszEntry[uEntrySize] = '\0';
	fclose(pLogFile);

	return pszEntry;
}

static int QueStatMessage(MessageQueue *pMQ, QueueMessage *pQM)
{
	/* Build the slog file path */
	char szSlogFilePath[SYS_MAX_PATH];

	QueGetFilePath(pMQ, pQM, szSlogFilePath, QUEUE_SLOG_DIR);

	FILE *pLogFile = fopen(szSlogFilePath, "a+t");

	if (pLogFile == NULL) {
		ErrSetErrorCode(ERR_FILE_OPEN, szSlogFilePath);
		return ERR_FILE_OPEN;
	}
	/* Dump peek time */
	time_t tCurr = time(NULL);
	char szTime[128];

	MscGetTimeStr(szTime, sizeof(szTime) - 1, tCurr);
	fprintf(pLogFile, "[PeekTime] %lu : %s\n", (unsigned long) tCurr, szTime);
	fclose(pLogFile);

	return 0;
}

QMSG_HANDLE QueCreateMessage(QUEUE_HANDLE hQueue)
{
	MessageQueue *pMQ = (MessageQueue *) hQueue;

	/* Build message file path */
	int iLevel1 = rand() % pMQ->iNumDirsLevel;
	int iLevel2 = rand() % pMQ->iNumDirsLevel;
	char szSubPath[SYS_MAX_PATH];
	char szMsgFilePath[SYS_MAX_PATH];

	SysSNPrintf(szSubPath, sizeof(szSubPath) - 1, "%s%d%s%d%s%s",
		    pMQ->pszRootPath, iLevel1, SYS_SLASH_STR,
		    iLevel2, SYS_SLASH_STR, QUEUE_TEMP_DIR);

	if (MscUniqueFile(szSubPath, szMsgFilePath, sizeof(szMsgFilePath)) < 0)
		return INVALID_QMSG_HANDLE;

	/* Extract file name */
	char szMsgFileName[SYS_MAX_PATH];

	MscGetFileName(szMsgFilePath, szMsgFileName);

	/* Create queue message data */
	QueueMessage *pQM = QueAllocMessage(iLevel1, iLevel2, QUEUE_TEMP_DIR,
					    szMsgFileName, 0, 0);

	if (pQM == NULL)
		return INVALID_QMSG_HANDLE;

	return (QMSG_HANDLE) pQM;
}

static int QueGetFilePath(MessageQueue *pMQ, QueueMessage *pQM, char *pszFilePath,
			  char const *pszQueueDir)
{
	if (pszQueueDir == NULL)
		pszQueueDir = pQM->pszQueueDir;

	SysSNPrintf(pszFilePath, SYS_MAX_PATH - 1, "%s%d%s%d%s%s%s%s",
		    pMQ->pszRootPath, pQM->iLevel1, SYS_SLASH_STR,
		    pQM->iLevel2, SYS_SLASH_STR, pszQueueDir, SYS_SLASH_STR, pQM->pszFileName);

	return 0;
}

int QueGetFilePath(QUEUE_HANDLE hQueue, QMSG_HANDLE hMessage, char *pszFilePath,
		   char const *pszQueueDir)
{
	MessageQueue *pMQ = (MessageQueue *) hQueue;
	QueueMessage *pQM = (QueueMessage *) hMessage;

	return QueGetFilePath(pMQ, pQM, pszFilePath, pszQueueDir);
}

static int QueDoMessageCleanup(QUEUE_HANDLE hQueue, QMSG_HANDLE hMessage)
{
	MessageQueue *pMQ = (MessageQueue *) hQueue;
	QueueMessage *pQM = (QueueMessage *) hMessage;
	char szQueueFilePath[SYS_MAX_PATH];

	if (pQM->ulFlags & QUMF_FREEZE) {
		/* Move message file */
		char szTargetFile[SYS_MAX_PATH];

		QueGetFilePath(pMQ, pQM, szQueueFilePath);
		QueGetFilePath(pMQ, pQM, szTargetFile, QUEUE_FROZ_DIR);
		if (SysMoveFile(szQueueFilePath, szTargetFile) < 0)
			return ErrGetErrorCode();

		/* Change message location */
		pQM->pszQueueDir = QUEUE_FROZ_DIR;
	} else {
		/* Clean message file */
		QueGetFilePath(pMQ, pQM, szQueueFilePath);
		SysRemove(szQueueFilePath);

		/* Clean 'info' file */
		QueGetFilePath(pMQ, pQM, szQueueFilePath, QUEUE_INFO_DIR);
		SysRemove(szQueueFilePath);

		/* Clean 'slog' file */
		QueGetFilePath(pMQ, pQM, szQueueFilePath, QUEUE_SLOG_DIR);
		SysRemove(szQueueFilePath);
	}

	/* Clean 'temp' file */
	QueGetFilePath(pMQ, pQM, szQueueFilePath, QUEUE_TEMP_DIR);
	SysRemove(szQueueFilePath);

	/* Clean 'cust' file */
	QueGetFilePath(pMQ, pQM, szQueueFilePath, QUEUE_CUST_DIR);
	SysRemove(szQueueFilePath);

	/* Clean 'mprc' file */
	QueGetFilePath(pMQ, pQM, szQueueFilePath, QUEUE_MPRC_DIR);
	SysRemove(szQueueFilePath);

	return 0;
}

int QueCloseMessage(QUEUE_HANDLE hQueue, QMSG_HANDLE hMessage)
{
	MessageQueue *pMQ = (MessageQueue *) hQueue;
	QueueMessage *pQM = (QueueMessage *) hMessage;

	if (pQM->ulFlags & QUMF_DELETED)
		QueDoMessageCleanup(hQueue, hMessage);
	QueFreeMessage(pQM);

	return 0;
}

QMSG_HANDLE QueGetHandle(QUEUE_HANDLE hQueue, int iLevel1, int iLevel2, char const *pszQueueDir,
			 char const *pszFileName)
{
	MessageQueue *pMQ = (MessageQueue *) hQueue;
	QueueMessage *pQM = QueAllocMessage(iLevel1, iLevel2, pszQueueDir, pszFileName, 0, 0);

	if (pQM == NULL)
		return INVALID_QMSG_HANDLE;

	/* Load message statistics */
	QueLoadMessageStat(pMQ, pQM);

	return (QMSG_HANDLE) pQM;
}

char const *QueGetFileName(QMSG_HANDLE hMessage)
{
	QueueMessage *pQM = (QueueMessage *) hMessage;

	return pQM->pszFileName;
}

char const *QueGetQueueDir(QMSG_HANDLE hMessage)
{
	QueueMessage *pQM = (QueueMessage *) hMessage;

	return pQM->pszQueueDir;
}

int QueGetLevel1(QMSG_HANDLE hMessage)
{
	QueueMessage *pQM = (QueueMessage *) hMessage;

	return pQM->iLevel1;
}

int QueGetLevel2(QMSG_HANDLE hMessage)
{
	QueueMessage *pQM = (QueueMessage *) hMessage;

	return pQM->iLevel2;
}

int QueGetTryCount(QMSG_HANDLE hMessage)
{
	QueueMessage *pQM = (QueueMessage *) hMessage;

	return pQM->iNumTries;
}

time_t QueGetLastTryTime(QMSG_HANDLE hMessage)
{
	QueueMessage *pQM = (QueueMessage *) hMessage;

	return pQM->tLastTry;
}

time_t QueGetMessageNextOp(QUEUE_HANDLE hQueue, QMSG_HANDLE hMessage)
{
	MessageQueue *pMQ = (MessageQueue *) hQueue;
	QueueMessage *pQM = (QueueMessage *) hMessage;

	return pQM->tLastTry + QueNextRetryOp(pQM->iNumTries, (unsigned int) pMQ->iRetryTimeout,
					      (unsigned int) pMQ->iRetryIncrRatio);
}

int QueInitMessageStats(QUEUE_HANDLE hQueue, QMSG_HANDLE hMessage)
{
	MessageQueue *pMQ = (MessageQueue *) hQueue;
	QueueMessage *pQM = (QueueMessage *) hMessage;
	char szQueueFilePath[SYS_MAX_PATH];

	QueGetFilePath(pMQ, pQM, szQueueFilePath, QUEUE_SLOG_DIR);
	SysRemove(szQueueFilePath);

	/* Init message statistics */
	pQM->iNumTries = 0;
	pQM->tLastTry = 0;

	return 0;
}

int QueCleanupMessage(QUEUE_HANDLE hQueue, QMSG_HANDLE hMessage, bool bFreeze)
{
	MessageQueue *pMQ = (MessageQueue *) hQueue;
	QueueMessage *pQM = (QueueMessage *) hMessage;

	pQM->ulFlags |= QUMF_DELETED;
	if (bFreeze)
		pQM->ulFlags |= QUMF_FREEZE;

	return 0;
}

static int QueAddNew(MessageQueue *pMQ, QueueMessage *pQM)
{
	/* Add the queue entry */
	if (SysLockMutex(pMQ->hMutex, SYS_INFINITE_TIMEOUT) < 0)
		return ErrGetErrorCode();

	SYS_LIST_ADDT(&pQM->LLink, &pMQ->ReadyQueue);
	++pMQ->iReadyCount;
	SysSetEvent(pMQ->hReadyEvent);
	SysUnlockMutex(pMQ->hMutex);

	return 0;
}

int QueCommitMessage(QUEUE_HANDLE hQueue, QMSG_HANDLE hMessage)
{
	MessageQueue *pMQ = (MessageQueue *) hQueue;
	QueueMessage *pQM = (QueueMessage *) hMessage;

	/* Move message file (if not in mess) */
	if (strcmp(pQM->pszQueueDir, QUEUE_MESS_DIR) != 0) {
		char szSourceFile[SYS_MAX_PATH];
		char szTargetFile[SYS_MAX_PATH];

		QueGetFilePath(pMQ, pQM, szSourceFile);
		QueGetFilePath(pMQ, pQM, szTargetFile, QUEUE_MESS_DIR);

		if (SysMoveFile(szSourceFile, szTargetFile) < 0)
			return ErrGetErrorCode();

		/* Change message location */
		pQM->pszQueueDir = QUEUE_MESS_DIR;
	}
	/* Unmask temporary flags */
	pQM->ulFlags = QUE_MASK_TMPFLAGS(pQM->ulFlags);

	/* Add to queue */
	if (QueAddNew(pMQ, pQM) < 0)
		return ErrGetErrorCode();

	return 0;
}

static bool QueMessageExpired(MessageQueue *pMQ, QueueMessage *pQM)
{
	return pQM->iNumTries >= pMQ->iMaxRetry;
}

static int QueAddRsnd(MessageQueue *pMQ, QueueMessage *pQM)
{
	/* Add the queue entry */
	if (SysLockMutex(pMQ->hMutex, SYS_INFINITE_TIMEOUT) < 0)
		return ErrGetErrorCode();

	SYS_LIST_ADDT(&pQM->LLink, &pMQ->RsndArenaQueue);
	++pMQ->iRsndArenaCount;
	SysUnlockMutex(pMQ->hMutex);

	return 0;
}

int QueResendMessage(QUEUE_HANDLE hQueue, QMSG_HANDLE hMessage)
{
	MessageQueue *pMQ = (MessageQueue *) hQueue;
	QueueMessage *pQM = (QueueMessage *) hMessage;

	/* Check for message expired */
	if (QueMessageExpired(pMQ, pQM)) {
		ErrSetErrorCode(ERR_SPOOL_FILE_EXPIRED);
		return ERR_SPOOL_FILE_EXPIRED;
	}
	/* Move message file (if not in rsnd dir) */
	if (strcmp(pQM->pszQueueDir, QUEUE_RSND_DIR) != 0) {
		char szSourceFile[SYS_MAX_PATH];
		char szTargetFile[SYS_MAX_PATH];

		QueGetFilePath(pMQ, pQM, szSourceFile);
		QueGetFilePath(pMQ, pQM, szTargetFile, QUEUE_RSND_DIR);

		if (SysMoveFile(szSourceFile, szTargetFile) < 0)
			return ErrGetErrorCode();

		/* Change message location */
		pQM->pszQueueDir = QUEUE_RSND_DIR;
	}
	/* Unmask temporary flags */
	pQM->ulFlags = QUE_MASK_TMPFLAGS(pQM->ulFlags);

	/* Add to queue */
	if (QueAddRsnd(pMQ, pQM) < 0)
		return ErrGetErrorCode();

	return 0;
}

QMSG_HANDLE QueExtractMessage(QUEUE_HANDLE hQueue, int iTimeout)
{
	MessageQueue *pMQ = (MessageQueue *) hQueue;

	/* Wait for message available */
	if (SysWaitEvent(pMQ->hReadyEvent, iTimeout) < 0)
		return INVALID_QMSG_HANDLE;

	if (SysLockMutex(pMQ->hMutex, SYS_INFINITE_TIMEOUT) < 0)
		return INVALID_QMSG_HANDLE;

	/* Get the first message of the queue */
	SysListHead *pLLink = SYS_LIST_FIRST(&pMQ->ReadyQueue);

	if (pLLink == NULL) {
		SysUnlockMutex(pMQ->hMutex);
		return INVALID_QMSG_HANDLE;
	}
	/* Remove the message from the list */
	SYS_LIST_DEL(pLLink);

	/* Decrement message count by resetting the event if no more messages are in */
	if (--pMQ->iReadyCount == 0)
		SysResetEvent(pMQ->hReadyEvent);

	SysUnlockMutex(pMQ->hMutex);

	/* Get queue message pointer */
	QueueMessage *pQM = SYS_LIST_ENTRY(pLLink, QueueMessage, LLink);

	/* Update message statistics */
	++pQM->iNumTries;
	pQM->tLastTry = time(NULL);

	/* Update log file */
	QueStatMessage(pMQ, pQM);

	return (QMSG_HANDLE) pQM;
}

int QueCheckMessage(QUEUE_HANDLE hQueue, QMSG_HANDLE hMessage)
{
	MessageQueue *pMQ = (MessageQueue *) hQueue;
	QueueMessage *pQM = (QueueMessage *) hMessage;
	char szQueueFilePath[SYS_MAX_PATH];

	if (pQM->ulFlags & QUMF_DELETED) {
		ErrSetErrorCode(ERR_MESSAGE_DELETED);
		return ERR_MESSAGE_DELETED;
	}

	QueGetFilePath(hQueue, hMessage, szQueueFilePath, QUEUE_MESS_DIR);

	if (!SysExistFile(szQueueFilePath)) {
		QueGetFilePath(hQueue, hMessage, szQueueFilePath, QUEUE_RSND_DIR);

		if (!SysExistFile(szQueueFilePath)) {
			ErrSetErrorCode(ERR_NO_MESSAGE_FILE);
			return ERR_NO_MESSAGE_FILE;
		}
	}

	return 0;
}

static bool QueMessageDestMatch(MessageQueue *pMQ, QueueMessage *pQM,
				char const *pszAddressMatch)
{
	SpoolFileHeader SFH;
	char szQueueFilePath[SYS_MAX_PATH];

	QueGetFilePath(pMQ, pQM, szQueueFilePath);
	if (USmlLoadSpoolFileHeader(szQueueFilePath, SFH) < 0)
		return false;

	bool bAddressMatch = false;

	if (strchr(pszAddressMatch, '@') == NULL) {
		/* RFC style ETRN (domain based) */
		char szDestUser[MAX_ADDR_NAME];
		char szDestDomain[MAX_ADDR_NAME];

		if (StrStringsCount(SFH.ppszRcpt) < 1 ||
		    USmtpSplitEmailAddr(SFH.ppszRcpt[0], szDestUser, szDestDomain) < 0) {
			USmlCleanupSpoolFileHeader(SFH);
			return false;
		}

		bAddressMatch = (StrIWildMatch(szDestDomain, pszAddressMatch) != 0);
	} else {
		/* XMail style ETRN (email based) */
		bAddressMatch = (StrIWildMatch(SFH.ppszRcpt[0], pszAddressMatch) != 0);
	}
	USmlCleanupSpoolFileHeader(SFH);

	return bAddressMatch;
}

int QueFlushRsndArena(QUEUE_HANDLE hQueue, char const *pszAddressMatch)
{
	MessageQueue *pMQ = (MessageQueue *) hQueue;

	if (SysLockMutex(pMQ->hMutex, SYS_INFINITE_TIMEOUT) < 0)
		return ErrGetErrorCode();

	SysListHead *pLLink;

	SYS_LIST_FOR_EACH(pLLink, &pMQ->RsndArenaQueue) {
		QueueMessage *pQM = SYS_LIST_ENTRY(pLLink, QueueMessage, LLink);

		if (pszAddressMatch == NULL ||
		    QueMessageDestMatch(pMQ, pQM, pszAddressMatch)) {
			/* Set the list pointer to the next item */
			pLLink = pLLink->pPrev;

			/* Remove item from resend arena */
			SYS_LIST_DEL(&pQM->LLink);
			--pMQ->iRsndArenaCount;

			/* Add item from resend queue */
			SYS_LIST_ADDT(&pQM->LLink, &pMQ->ReadyQueue);
			++pMQ->iReadyCount;
		}
	}

	/* If the count of rsnd queue is not zero, set the event */
	if (pMQ->iReadyCount > 0)
		SysSetEvent(pMQ->hReadyEvent);
	SysUnlockMutex(pMQ->hMutex);

	return 0;
}