github/OpenArchiver
1
1
Fork 0
mirror of https://github.com/LogicLabs-OU/OpenArchiver.git synced 2026-04-06 00:31:57 +02:00
Code Issues 161 Packages Projects Releases Wiki Activity

Formatting code

Browse Source
This commit is contained in:
Wayne
2025-10-23 17:01:41 +02:00
parent b7799f749d
commit 8ff772fba2
60 changed files with 6504 additions and 6759 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
.github/ISSUE_TEMPLATE/bug_report.md vendored
View File

@@ -4,7 +4,6 @@ about: Create a report to help us improve
title: ''
labels: bug
assignees: ''
---
**Describe the bug**
@@ -12,9 +11,10 @@ A clear and concise description of what the bug is.
**To Reproduce**
Steps to reproduce the behavior:
1. Go to '...'
2. Click on '....'
5. See error
3. See error
**Expected behavior**
A clear and concise description of what you expected to happen.
@@ -23,7 +23,8 @@ A clear and concise description of what you expected to happen.
If applicable, add screenshots to help explain your problem.
**System:**
- Open Archiver Version:
- Open Archiver Version:
**Relevant logs:**
Any relevant logs (Redact sensitive information)

3
.github/ISSUE_TEMPLATE/feature_request.md vendored
View File

@@ -4,11 +4,10 @@ about: Suggest an idea for this project
title: ''
labels: enhancement
assignees: ''
---
**Is your feature request related to a problem? Please describe.**
A clear and concise description of what the problem is.
A clear and concise description of what the problem is.
**Describe the solution you'd like**
A clear and concise description of what you want to happen.

28
apps/open-archiver/index.ts
View File

@@ -4,23 +4,21 @@ import * as dotenv from 'dotenv';
dotenv.config();
async function start() {
// --- Environment Variable Validation ---
const { PORT_BACKEND } = process.env;
// --- Environment Variable Validation ---
const { PORT_BACKEND } = process.env;
if (!PORT_BACKEND) {
throw new Error(
'Missing required environment variables for the backend: PORT_BACKEND.'
);
}
// Create the server instance (passing no modules for the default OSS version)
const app = await createServer([]);
if (!PORT_BACKEND) {
throw new Error('Missing required environment variables for the backend: PORT_BACKEND.');
}
// Create the server instance (passing no modules for the default OSS version)
const app = await createServer([]);
app.listen(PORT_BACKEND, () => {
logger.info({}, `✅ Open Archiver (OSS) running on port ${PORT_BACKEND}`);
});
app.listen(PORT_BACKEND, () => {
logger.info({}, `✅ Open Archiver (OSS) running on port ${PORT_BACKEND}`);
});
}
start().catch(error => {
logger.error({ error }, 'Failed to start the server:', error);
process.exit(1);
start().catch((error) => {
logger.error({ error }, 'Failed to start the server:', error);
process.exit(1);
});

20
packages/backend/src/api/controllers/ingestion.controller.ts
View File

@@ -24,7 +24,6 @@ export class IngestionController {
}
public create = async (req: Request, res: Response): Promise<Response> => {
try {
const dto: CreateIngestionSourceDto = req.body;
const userId = req.user?.sub;
@@ -35,7 +34,12 @@ export class IngestionController {
if (!actor) {
return res.status(401).json({ message: req.t('errors.unauthorized') });
}
const newSource = await IngestionService.create(dto, userId, actor, req.ip || 'unknown');
const newSource = await IngestionService.create(
dto,
userId,
actor,
req.ip || 'unknown'
);
const safeSource = this.toSafeIngestionSource(newSource);
return res.status(201).json(safeSource);
} catch (error: any) {
@@ -78,7 +82,6 @@ export class IngestionController {
};
public update = async (req: Request, res: Response): Promise<Response> => {
try {
const { id } = req.params;
const dto: UpdateIngestionSourceDto = req.body;
@@ -90,7 +93,12 @@ export class IngestionController {
if (!actor) {
return res.status(401).json({ message: req.t('errors.unauthorized') });
}
const updatedSource = await IngestionService.update(id, dto, actor, req.ip || 'unknown');
const updatedSource = await IngestionService.update(
id,
dto,
actor,
req.ip || 'unknown'
);
const safeSource = this.toSafeIngestionSource(updatedSource);
return res.status(200).json(safeSource);
} catch (error) {
@@ -103,7 +111,6 @@ export class IngestionController {
};
public delete = async (req: Request, res: Response): Promise<Response> => {
try {
checkDeletionEnabled();
const { id } = req.params;
@@ -129,7 +136,6 @@ export class IngestionController {
};
public triggerInitialImport = async (req: Request, res: Response): Promise<Response> => {
try {
const { id } = req.params;
await IngestionService.triggerInitialImport(id);
@@ -144,7 +150,6 @@ export class IngestionController {
};
public pause = async (req: Request, res: Response): Promise<Response> => {
try {
const { id } = req.params;
const userId = req.user?.sub;
@@ -173,7 +178,6 @@ export class IngestionController {
};
public triggerForceSync = async (req: Request, res: Response): Promise<Response> => {
try {
const { id } = req.params;
const userId = req.user?.sub;

38
packages/backend/src/api/controllers/integrity.controller.ts
View File

@@ -3,27 +3,27 @@ import { IntegrityService } from '../../services/IntegrityService';
import { z } from 'zod';
const checkIntegritySchema = z.object({
id: z.string().uuid(),
id: z.string().uuid(),
});
export class IntegrityController {
private integrityService = new IntegrityService();
private integrityService = new IntegrityService();
public checkIntegrity = async (req: Request, res: Response) => {
try {
const { id } = checkIntegritySchema.parse(req.params);
const results = await this.integrityService.checkEmailIntegrity(id);
res.status(200).json(results);
} catch (error) {
if (error instanceof z.ZodError) {
return res
.status(400)
.json({ message: req.t('api.requestBodyInvalid'), errors: error.message });
}
if (error instanceof Error && error.message === 'Archived email not found') {
return res.status(404).json({ message: req.t('errors.notFound') });
}
res.status(500).json({ message: req.t('errors.internalServerError') });
}
};
public checkIntegrity = async (req: Request, res: Response) => {
try {
const { id } = checkIntegritySchema.parse(req.params);
const results = await this.integrityService.checkEmailIntegrity(id);
res.status(200).json(results);
} catch (error) {
if (error instanceof z.ZodError) {
return res
.status(400)
.json({ message: req.t('api.requestBodyInvalid'), errors: error.message });
}
if (error instanceof Error && error.message === 'Archived email not found') {
return res.status(404).json({ message: req.t('errors.notFound') });
}
res.status(500).json({ message: req.t('errors.internalServerError') });
}
};
}

64
packages/backend/src/api/controllers/jobs.controller.ts
View File

@@ -1,42 +1,42 @@
import { Request, Response } from 'express';
import { JobsService } from '../../services/JobsService';
import {
IGetQueueJobsRequestParams,
IGetQueueJobsRequestQuery,
JobStatus,
IGetQueueJobsRequestParams,
IGetQueueJobsRequestQuery,
JobStatus,
} from '@open-archiver/types';
export class JobsController {
private jobsService: JobsService;
private jobsService: JobsService;
constructor() {
this.jobsService = new JobsService();
}
constructor() {
this.jobsService = new JobsService();
}
public getQueues = async (req: Request, res: Response) => {
try {
const queues = await this.jobsService.getQueues();
res.status(200).json({ queues });
} catch (error) {
res.status(500).json({ message: 'Error fetching queues', error });
}
};
public getQueues = async (req: Request, res: Response) => {
try {
const queues = await this.jobsService.getQueues();
res.status(200).json({ queues });
} catch (error) {
res.status(500).json({ message: 'Error fetching queues', error });
}
};
public getQueueJobs = async (req: Request, res: Response) => {
try {
const { queueName } = req.params as unknown as IGetQueueJobsRequestParams;
const { status, page, limit } = req.query as unknown as IGetQueueJobsRequestQuery;
const pageNumber = parseInt(page, 10) || 1;
const limitNumber = parseInt(limit, 10) || 10;
const queueDetails = await this.jobsService.getQueueDetails(
queueName,
status,
pageNumber,
limitNumber
);
res.status(200).json(queueDetails);
} catch (error) {
res.status(500).json({ message: 'Error fetching queue jobs', error });
}
};
public getQueueJobs = async (req: Request, res: Response) => {
try {
const { queueName } = req.params as unknown as IGetQueueJobsRequestParams;
const { status, page, limit } = req.query as unknown as IGetQueueJobsRequestQuery;
const pageNumber = parseInt(page, 10) || 1;
const limitNumber = parseInt(limit, 10) || 10;
const queueDetails = await this.jobsService.getQueueDetails(
queueName,
status,
pageNumber,
limitNumber
);
res.status(200).json(queueDetails);
} catch (error) {
res.status(500).json({ message: 'Error fetching queue jobs', error });
}
};
}

1
packages/backend/src/api/controllers/user.controller.ts
View File

@@ -4,7 +4,6 @@ import * as schema from '../../database/schema';
import { sql } from 'drizzle-orm';
import { db } from '../../database';
const userService = new UserService();
export const getUsers = async (req: Request, res: Response) => {

10
packages/backend/src/api/routes/integrity.routes.ts
View File

@@ -5,12 +5,12 @@ import { requirePermission } from '../middleware/requirePermission';
import { AuthService } from '../../services/AuthService';
export const integrityRoutes = (authService: AuthService): Router => {
const router = Router();
const controller = new IntegrityController();
const router = Router();
const controller = new IntegrityController();
router.use(requireAuth(authService));
router.use(requireAuth(authService));
router.get('/:id', requirePermission('read', 'archive'), controller.checkIntegrity);
router.get('/:id', requirePermission('read', 'archive'), controller.checkIntegrity);
return router;
return router;
};

28
packages/backend/src/api/routes/jobs.routes.ts
View File

@@ -5,21 +5,21 @@ import { requirePermission } from '../middleware/requirePermission';
import { AuthService } from '../../services/AuthService';
export const createJobsRouter = (authService: AuthService): Router => {
const router = Router();
const jobsController = new JobsController();
const router = Router();
const jobsController = new JobsController();
router.use(requireAuth(authService));
router.use(requireAuth(authService));
router.get(
'/queues',
requirePermission('manage', 'all', 'user.requiresSuperAdminRole'),
jobsController.getQueues
);
router.get(
'/queues/:queueName',
requirePermission('manage', 'all', 'user.requiresSuperAdminRole'),
jobsController.getQueueJobs
);
router.get(
'/queues',
requirePermission('manage', 'all', 'user.requiresSuperAdminRole'),
jobsController.getQueues
);
router.get(
'/queues/:queueName',
requirePermission('manage', 'all', 'user.requiresSuperAdminRole'),
jobsController.getQueueJobs
);
return router;
return router;
};

218
packages/backend/src/api/server.ts
View File

@@ -37,134 +37,134 @@ import { config } from '../config';
import { OpenArchiverFeature } from '@open-archiver/types';
// Define the "plugin" interface
export interface ArchiverModule {
initialize: (app: Express, authService: AuthService) => Promise<void>;
name: OpenArchiverFeature;
initialize: (app: Express, authService: AuthService) => Promise<void>;
name: OpenArchiverFeature;
}
export let authService: AuthService;
export async function createServer(modules: ArchiverModule[] = []): Promise<Express> {
// Load environment variables
dotenv.config();
// Load environment variables
dotenv.config();
// --- Environment Variable Validation ---
const { JWT_SECRET, JWT_EXPIRES_IN } = process.env;
// --- Environment Variable Validation ---
const { JWT_SECRET, JWT_EXPIRES_IN } = process.env;
if (!JWT_SECRET || !JWT_EXPIRES_IN) {
throw new Error(
'Missing required environment variables for the backend: JWT_SECRET, JWT_EXPIRES_IN.'
);
}
if (!JWT_SECRET || !JWT_EXPIRES_IN) {
throw new Error(
'Missing required environment variables for the backend: JWT_SECRET, JWT_EXPIRES_IN.'
);
}
// --- Dependency Injection Setup ---
const auditService = new AuditService();
const userService = new UserService();
authService = new AuthService(userService, auditService, JWT_SECRET, JWT_EXPIRES_IN);
const authController = new AuthController(authService, userService);
const ingestionController = new IngestionController();
const archivedEmailController = new ArchivedEmailController();
const storageService = new StorageService();
const storageController = new StorageController(storageService);
const searchService = new SearchService();
const searchController = new SearchController();
const iamService = new IamService();
const iamController = new IamController(iamService);
const settingsService = new SettingsService();
// --- Dependency Injection Setup ---
const auditService = new AuditService();
const userService = new UserService();
authService = new AuthService(userService, auditService, JWT_SECRET, JWT_EXPIRES_IN);
const authController = new AuthController(authService, userService);
const ingestionController = new IngestionController();
const archivedEmailController = new ArchivedEmailController();
const storageService = new StorageService();
const storageController = new StorageController(storageService);
const searchService = new SearchService();
const searchController = new SearchController();
const iamService = new IamService();
const iamController = new IamController(iamService);
const settingsService = new SettingsService();
// --- i18next Initialization ---
const initializeI18next = async () => {
const systemSettings = await settingsService.getSystemSettings();
const defaultLanguage = systemSettings?.language || 'en';
logger.info({ language: defaultLanguage }, 'Default language');
await i18next.use(FsBackend).init({
lng: defaultLanguage,
fallbackLng: defaultLanguage,
ns: ['translation'],
defaultNS: 'translation',
backend: {
loadPath: path.resolve(__dirname, '../locales/{{lng}}/{{ns}}.json'),
},
});
};
// --- i18next Initialization ---
const initializeI18next = async () => {
const systemSettings = await settingsService.getSystemSettings();
const defaultLanguage = systemSettings?.language || 'en';
logger.info({ language: defaultLanguage }, 'Default language');
await i18next.use(FsBackend).init({
lng: defaultLanguage,
fallbackLng: defaultLanguage,
ns: ['translation'],
defaultNS: 'translation',
backend: {
loadPath: path.resolve(__dirname, '../locales/{{lng}}/{{ns}}.json'),
},
});
};
// Initialize i18next
await initializeI18next();
logger.info({}, 'i18next initialized');
// Initialize i18next
await initializeI18next();
logger.info({}, 'i18next initialized');
// Configure the Meilisearch index on startup
logger.info({}, 'Configuring email index...');
await searchService.configureEmailIndex();
// Configure the Meilisearch index on startup
logger.info({}, 'Configuring email index...');
await searchService.configureEmailIndex();
const app = express();
const app = express();
// --- CORS ---
app.use(
cors({
origin: process.env.APP_URL || 'http://localhost:3000',
credentials: true,
})
);
// --- CORS ---
app.use(
cors({
origin: process.env.APP_URL || 'http://localhost:3000',
credentials: true,
})
);
// Trust the proxy to get the real IP address of the client.
// This is important for audit logging and security.
app.set('trust proxy', true);
// Trust the proxy to get the real IP address of the client.
// This is important for audit logging and security.
app.set('trust proxy', true);
// --- Routes ---
const authRouter = createAuthRouter(authController);
const ingestionRouter = createIngestionRouter(ingestionController, authService);
const archivedEmailRouter = createArchivedEmailRouter(archivedEmailController, authService);
const storageRouter = createStorageRouter(storageController, authService);
const searchRouter = createSearchRouter(searchController, authService);
const dashboardRouter = createDashboardRouter(authService);
const iamRouter = createIamRouter(iamController, authService);
const uploadRouter = createUploadRouter(authService);
const userRouter = createUserRouter(authService);
const settingsRouter = createSettingsRouter(authService);
const apiKeyRouter = apiKeyRoutes(authService);
const integrityRouter = integrityRoutes(authService);
const jobsRouter = createJobsRouter(authService);
// --- Routes ---
const authRouter = createAuthRouter(authController);
const ingestionRouter = createIngestionRouter(ingestionController, authService);
const archivedEmailRouter = createArchivedEmailRouter(archivedEmailController, authService);
const storageRouter = createStorageRouter(storageController, authService);
const searchRouter = createSearchRouter(searchController, authService);
const dashboardRouter = createDashboardRouter(authService);
const iamRouter = createIamRouter(iamController, authService);
const uploadRouter = createUploadRouter(authService);
const userRouter = createUserRouter(authService);
const settingsRouter = createSettingsRouter(authService);
const apiKeyRouter = apiKeyRoutes(authService);
const integrityRouter = integrityRoutes(authService);
const jobsRouter = createJobsRouter(authService);
// Middleware for all other routes
app.use((req, res, next) => {
// exclude certain API endpoints from the rate limiter, for example status, system settings
const excludedPatterns = [/^\/v\d+\/auth\/status$/, /^\/v\d+\/settings\/system$/];
for (const pattern of excludedPatterns) {
if (pattern.test(req.path)) {
return next();
}
}
rateLimiter(req, res, next);
});
app.use(express.json());
app.use(express.urlencoded({ extended: true }));
// Middleware for all other routes
app.use((req, res, next) => {
// exclude certain API endpoints from the rate limiter, for example status, system settings
const excludedPatterns = [/^\/v\d+\/auth\/status$/, /^\/v\d+\/settings\/system$/];
for (const pattern of excludedPatterns) {
if (pattern.test(req.path)) {
return next();
}
}
rateLimiter(req, res, next);
});
app.use(express.json());
app.use(express.urlencoded({ extended: true }));
// i18n middleware
app.use(i18nextMiddleware.handle(i18next));
// i18n middleware
app.use(i18nextMiddleware.handle(i18next));
app.use(`/${config.api.version}/auth`, authRouter);
app.use(`/${config.api.version}/iam`, iamRouter);
app.use(`/${config.api.version}/upload`, uploadRouter);
app.use(`/${config.api.version}/ingestion-sources`, ingestionRouter);
app.use(`/${config.api.version}/archived-emails`, archivedEmailRouter);
app.use(`/${config.api.version}/storage`, storageRouter);
app.use(`/${config.api.version}/search`, searchRouter);
app.use(`/${config.api.version}/dashboard`, dashboardRouter);
app.use(`/${config.api.version}/users`, userRouter);
app.use(`/${config.api.version}/settings`, settingsRouter);
app.use(`/${config.api.version}/api-keys`, apiKeyRouter);
app.use(`/${config.api.version}/integrity`, integrityRouter);
app.use(`/${config.api.version}/jobs`, jobsRouter);
app.use(`/${config.api.version}/auth`, authRouter);
app.use(`/${config.api.version}/iam`, iamRouter);
app.use(`/${config.api.version}/upload`, uploadRouter);
app.use(`/${config.api.version}/ingestion-sources`, ingestionRouter);
app.use(`/${config.api.version}/archived-emails`, archivedEmailRouter);
app.use(`/${config.api.version}/storage`, storageRouter);
app.use(`/${config.api.version}/search`, searchRouter);
app.use(`/${config.api.version}/dashboard`, dashboardRouter);
app.use(`/${config.api.version}/users`, userRouter);
app.use(`/${config.api.version}/settings`, settingsRouter);
app.use(`/${config.api.version}/api-keys`, apiKeyRouter);
app.use(`/${config.api.version}/integrity`, integrityRouter);
app.use(`/${config.api.version}/jobs`, jobsRouter);
// Load all provided extension modules
for (const module of modules) {
await module.initialize(app, authService);
console.log(`🏢 Enterprise module loaded: ${module.name}`);
}
app.get('/', (req, res) => {
res.send('Backend is running!!');
});
// Load all provided extension modules
for (const module of modules) {
await module.initialize(app, authService);
console.log(`🏢 Enterprise module loaded: ${module.name}`);
}
app.get('/', (req, res) => {
res.send('Backend is running!!');
});
console.log('✅ Core OSS modules loaded.');
console.log('✅ Core OSS modules loaded.');
return app;
return app;
}

2
packages/backend/src/config/api.ts
View File

@@ -9,5 +9,5 @@ export const apiConfig = {
? parseInt(process.env.RATE_LIMIT_MAX_REQUESTS, 10)
: 100, // limit each IP to 100 requests per windowMs
},
version: 'v1'
version: 'v1',
};

2421
packages/backend/src/database/migrations/meta/0020_snapshot.json
View File

File diff suppressed because it is too large Load Diff

2515
packages/backend/src/database/migrations/meta/0021_snapshot.json
View File

File diff suppressed because it is too large Load Diff

2581
packages/backend/src/database/migrations/meta/0022_snapshot.json
View File

File diff suppressed because it is too large Load Diff

2581
packages/backend/src/database/migrations/meta/0023_snapshot.json
View File

File diff suppressed because it is too large Load Diff

346
packages/backend/src/database/migrations/meta/_journal.json
View File

@@ -1,174 +1,174 @@
{
"version": "7",
"dialect": "postgresql",
"entries": [
{
"idx": 0,
"version": "7",
"when": 1752225352591,
"tag": "0000_amusing_namora",
"breakpoints": true
},
{
"idx": 1,
"version": "7",
"when": 1752326803882,
"tag": "0001_odd_night_thrasher",
"breakpoints": true
},
{
"idx": 2,
"version": "7",
"when": 1752332648392,
"tag": "0002_lethal_quentin_quire",
"breakpoints": true
},
{
"idx": 3,
"version": "7",
"when": 1752332967084,
"tag": "0003_petite_wrecker",
"breakpoints": true
},
{
"idx": 4,
"version": "7",
"when": 1752606108876,
"tag": "0004_sleepy_paper_doll",
"breakpoints": true
},
{
"idx": 5,
"version": "7",
"when": 1752606327253,
"tag": "0005_chunky_sue_storm",
"breakpoints": true
},
{
"idx": 6,
"version": "7",
"when": 1753112018514,
"tag": "0006_majestic_caretaker",
"breakpoints": true
},
{
"idx": 7,
"version": "7",
"when": 1753190159356,
"tag": "0007_handy_archangel",
"breakpoints": true
},
{
"idx": 8,
"version": "7",
"when": 1753370737317,
"tag": "0008_eminent_the_spike",
"breakpoints": true
},
{
"idx": 9,
"version": "7",
"when": 1754337938241,
"tag": "0009_late_lenny_balinger",
"breakpoints": true
},
{
"idx": 10,
"version": "7",
"when": 1754420780849,
"tag": "0010_perpetual_lightspeed",
"breakpoints": true
},
{
"idx": 11,
"version": "7",
"when": 1754422064158,
"tag": "0011_tan_blackheart",
"breakpoints": true
},
{
"idx": 12,
"version": "7",
"when": 1754476962901,
"tag": "0012_warm_the_stranger",
"breakpoints": true
},
{
"idx": 13,
"version": "7",
"when": 1754659373517,
"tag": "0013_classy_talkback",
"breakpoints": true
},
{
"idx": 14,
"version": "7",
"when": 1754831765718,
"tag": "0014_foamy_vapor",
"breakpoints": true
},
{
"idx": 15,
"version": "7",
"when": 1755443936046,
"tag": "0015_wakeful_norman_osborn",
"breakpoints": true
},
{
"idx": 16,
"version": "7",
"when": 1755780572342,
"tag": "0016_lonely_mariko_yashida",
"breakpoints": true
},
{
"idx": 17,
"version": "7",
"when": 1755961566627,
"tag": "0017_tranquil_shooting_star",
"breakpoints": true
},
{
"idx": 18,
"version": "7",
"when": 1756911118035,
"tag": "0018_flawless_owl",
"breakpoints": true
},
{
"idx": 19,
"version": "7",
"when": 1756937533843,
"tag": "0019_confused_scream",
"breakpoints": true
},
{
"idx": 20,
"version": "7",
"when": 1757860242528,
"tag": "0020_panoramic_wolverine",
"breakpoints": true
},
{
"idx": 21,
"version": "7",
"when": 1759412986134,
"tag": "0021_nosy_veda",
"breakpoints": true
},
{
"idx": 22,
"version": "7",
"when": 1759701622932,
"tag": "0022_complete_triton",
"breakpoints": true
},
{
"idx": 23,
"version": "7",
"when": 1760354094610,
"tag": "0023_swift_swordsman",
"breakpoints": true
}
]
}
"version": "7",
"dialect": "postgresql",
"entries": [
{
"idx": 0,
"version": "7",
"when": 1752225352591,
"tag": "0000_amusing_namora",
"breakpoints": true
},
{
"idx": 1,
"version": "7",
"when": 1752326803882,
"tag": "0001_odd_night_thrasher",
"breakpoints": true
},
{
"idx": 2,
"version": "7",
"when": 1752332648392,
"tag": "0002_lethal_quentin_quire",
"breakpoints": true
},
{
"idx": 3,
"version": "7",
"when": 1752332967084,
"tag": "0003_petite_wrecker",
"breakpoints": true
},
{
"idx": 4,
"version": "7",
"when": 1752606108876,
"tag": "0004_sleepy_paper_doll",
"breakpoints": true
},
{
"idx": 5,
"version": "7",
"when": 1752606327253,
"tag": "0005_chunky_sue_storm",
"breakpoints": true
},
{
"idx": 6,
"version": "7",
"when": 1753112018514,
"tag": "0006_majestic_caretaker",
"breakpoints": true
},
{
"idx": 7,
"version": "7",
"when": 1753190159356,
"tag": "0007_handy_archangel",
"breakpoints": true
},
{
"idx": 8,
"version": "7",
"when": 1753370737317,
"tag": "0008_eminent_the_spike",
"breakpoints": true
},
{
"idx": 9,
"version": "7",
"when": 1754337938241,
"tag": "0009_late_lenny_balinger",
"breakpoints": true
},
{
"idx": 10,
"version": "7",
"when": 1754420780849,
"tag": "0010_perpetual_lightspeed",
"breakpoints": true
},
{
"idx": 11,
"version": "7",
"when": 1754422064158,
"tag": "0011_tan_blackheart",
"breakpoints": true
},
{
"idx": 12,
"version": "7",
"when": 1754476962901,
"tag": "0012_warm_the_stranger",
"breakpoints": true
},
{
"idx": 13,
"version": "7",
"when": 1754659373517,
"tag": "0013_classy_talkback",
"breakpoints": true
},
{
"idx": 14,
"version": "7",
"when": 1754831765718,
"tag": "0014_foamy_vapor",
"breakpoints": true
},
{
"idx": 15,
"version": "7",
"when": 1755443936046,
"tag": "0015_wakeful_norman_osborn",
"breakpoints": true
},
{
"idx": 16,
"version": "7",
"when": 1755780572342,
"tag": "0016_lonely_mariko_yashida",
"breakpoints": true
},
{
"idx": 17,
"version": "7",
"when": 1755961566627,
"tag": "0017_tranquil_shooting_star",
"breakpoints": true
},
{
"idx": 18,
"version": "7",
"when": 1756911118035,
"tag": "0018_flawless_owl",
"breakpoints": true
},
{
"idx": 19,
"version": "7",
"when": 1756937533843,
"tag": "0019_confused_scream",
"breakpoints": true
},
{
"idx": 20,
"version": "7",
"when": 1757860242528,
"tag": "0020_panoramic_wolverine",
"breakpoints": true
},
{
"idx": 21,
"version": "7",
"when": 1759412986134,
"tag": "0021_nosy_veda",
"breakpoints": true
},
{
"idx": 22,
"version": "7",
"when": 1759701622932,
"tag": "0022_complete_triton",
"breakpoints": true
},
{
"idx": 23,
"version": "7",
"when": 1760354094610,
"tag": "0023_swift_swordsman",
"breakpoints": true
}
]
}

4
packages/backend/src/database/schema/attachments.ts
View File

@@ -16,9 +16,7 @@ export const attachments = pgTable(
onDelete: 'cascade',
}),
},
(table) => [
index('source_hash_idx').on(table.ingestionSourceId, table.contentHashSha256),
]
(table) => [index('source_hash_idx').on(table.ingestionSourceId, table.contentHashSha256)]
);
export const emailAttachments = pgTable(

8
packages/backend/src/helpers/deletionGuard.ts
View File

@@ -2,8 +2,8 @@ import { config } from '../config';
import i18next from 'i18next';
export function checkDeletionEnabled() {
if (!config.app.enableDeletion) {
const errorMessage = i18next.t('Deletion is disabled for this instance.');
throw new Error(errorMessage);
}
if (!config.app.enableDeletion) {
const errorMessage = i18next.t('Deletion is disabled for this instance.');
throw new Error(errorMessage);
}
}

17
packages/backend/src/helpers/textExtractor.ts
View File

@@ -59,14 +59,17 @@ async function extractTextLegacy(buffer: Buffer, mimeType: string): Promise<stri
try {
if (mimeType === 'application/pdf') {
// Check PDF size (memory protection)
if (buffer.length > 50 * 1024 * 1024) { // 50MB Limit
if (buffer.length > 50 * 1024 * 1024) {
// 50MB Limit
logger.warn('PDF too large for legacy extraction, skipping');
return '';
}
return await extractTextFromPdf(buffer);
}
if (mimeType === 'application/vnd.openxmlformats-officedocument.wordprocessingml.document') {
if (
mimeType === 'application/vnd.openxmlformats-officedocument.wordprocessingml.document'
) {
const { value } = await mammoth.extractRawText({ buffer });
return value;
}
@@ -118,7 +121,9 @@ export async function extractText(buffer: Buffer, mimeType: string): Promise<str
// General size limit
const maxSize = process.env.TIKA_URL ? 100 * 1024 * 1024 : 50 * 1024 * 1024; // 100MB for Tika, 50MB for Legacy
if (buffer.length > maxSize) {
logger.warn(`File too large for text extraction: ${buffer.length} bytes (limit: ${maxSize})`);
logger.warn(
`File too large for text extraction: ${buffer.length} bytes (limit: ${maxSize})`
);
return '';
}
@@ -128,12 +133,12 @@ export async function extractText(buffer: Buffer, mimeType: string): Promise<str
if (tikaUrl) {
// Tika decides what it can parse
logger.debug(`Using Tika for text extraction: ${mimeType}`);
const ocrService = new OcrService()
const ocrService = new OcrService();
try {
return await ocrService.extractTextWithTika(buffer, mimeType);
} catch (error) {
logger.error({ error }, "OCR text extraction failed, returning empty string")
return ''
logger.error({ error }, 'OCR text extraction failed, returning empty string');
return '';
}
} else {
// extract using legacy mode

2
packages/backend/src/index.ts
View File

@@ -4,7 +4,7 @@ export { config } from './config';
export * from './services/AuthService';
export * from './services/AuditService';
export * from './api/middleware/requireAuth';
export * from './api/middleware/requirePermission'
export * from './api/middleware/requirePermission';
export { db } from './database';
export * as drizzleOrm from 'drizzle-orm';
export * from './database/schema';

6
packages/backend/src/jobs/processors/index-email-batch.processor.ts
View File

@@ -11,7 +11,7 @@ const databaseService = new DatabaseService();
const indexingService = new IndexingService(databaseService, searchService, storageService);
export default async function (job: Job<{ emails: PendingEmail[] }>) {
const { emails } = job.data;
console.log(`Indexing email batch with ${emails.length} emails`);
await indexingService.indexEmailBatch(emails);
const { emails } = job.data;
console.log(`Indexing email batch with ${emails.length} emails`);
await indexingService.indexEmailBatch(emails);
}

1
packages/backend/src/jobs/processors/process-mailbox.processor.ts
View File

@@ -15,7 +15,6 @@ import { DatabaseService } from '../../services/DatabaseService';
import { config } from '../../config';
import { indexingQueue } from '../queues';
/**
* This processor handles the ingestion of emails for a single user's mailbox.
* If an error occurs during processing (e.g., an API failure),

324
packages/backend/src/services/AuditService.ts
View File

@@ -1,193 +1,199 @@
import { db, Database } from '../database';
import * as schema from '../database/schema';
import { AuditLogEntry, CreateAuditLogEntry, GetAuditLogsOptions, GetAuditLogsResponse } from '@open-archiver/types';
import {
AuditLogEntry,
CreateAuditLogEntry,
GetAuditLogsOptions,
GetAuditLogsResponse,
} from '@open-archiver/types';
import { desc, sql, asc, and, gte, lte, eq } from 'drizzle-orm';
import { createHash } from 'crypto';
export class AuditService {
private sanitizeObject(obj: any): any {
if (obj === null || typeof obj !== 'object') {
return obj;
}
if (Array.isArray(obj)) {
return obj.map((item) => this.sanitizeObject(item));
}
const sanitizedObj: { [key: string]: any } = {};
for (const key in obj) {
if (Object.prototype.hasOwnProperty.call(obj, key)) {
const value = obj[key];
sanitizedObj[key] = value === undefined ? null : this.sanitizeObject(value);
}
}
return sanitizedObj;
}
private sanitizeObject(obj: any): any {
if (obj === null || typeof obj !== 'object') {
return obj;
}
if (Array.isArray(obj)) {
return obj.map((item) => this.sanitizeObject(item));
}
const sanitizedObj: { [key: string]: any } = {};
for (const key in obj) {
if (Object.prototype.hasOwnProperty.call(obj, key)) {
const value = obj[key];
sanitizedObj[key] = value === undefined ? null : this.sanitizeObject(value);
}
}
return sanitizedObj;
}
public async createAuditLog(entry: CreateAuditLogEntry) {
return db.transaction(async (tx) => {
// Lock the table to prevent race conditions
await tx.execute(sql`LOCK TABLE audit_logs IN EXCLUSIVE MODE`);
public async createAuditLog(entry: CreateAuditLogEntry) {
return db.transaction(async (tx) => {
// Lock the table to prevent race conditions
await tx.execute(sql`LOCK TABLE audit_logs IN EXCLUSIVE MODE`);
const sanitizedEntry = this.sanitizeObject(entry);
const sanitizedEntry = this.sanitizeObject(entry);
const previousHash = await this.getLatestHash(tx);
const newEntry = {
...sanitizedEntry,
previousHash,
timestamp: new Date()
};
const currentHash = this.calculateHash(newEntry);
const previousHash = await this.getLatestHash(tx);
const newEntry = {
...sanitizedEntry,
previousHash,
timestamp: new Date(),
};
const currentHash = this.calculateHash(newEntry);
const finalEntry = {
...newEntry,
currentHash
};
const finalEntry = {
...newEntry,
currentHash,
};
await tx.insert(schema.auditLogs).values(finalEntry);
await tx.insert(schema.auditLogs).values(finalEntry);
return finalEntry;
});
}
return finalEntry;
});
}
private async getLatestHash(tx: Database): Promise<string | null> {
const [latest] = await tx
.select({
currentHash: schema.auditLogs.currentHash
})
.from(schema.auditLogs)
.orderBy(desc(schema.auditLogs.id))
.limit(1);
private async getLatestHash(tx: Database): Promise<string | null> {
const [latest] = await tx
.select({
currentHash: schema.auditLogs.currentHash,
})
.from(schema.auditLogs)
.orderBy(desc(schema.auditLogs.id))
.limit(1);
return latest?.currentHash ?? null;
}
return latest?.currentHash ?? null;
}
private calculateHash(entry: any): string {
// Create a canonical object for hashing to ensure consistency in property order and types.
const objectToHash = {
actorIdentifier: entry.actorIdentifier,
actorIp: entry.actorIp ?? null,
actionType: entry.actionType,
targetType: entry.targetType ?? null,
targetId: entry.targetId ?? null,
details: entry.details ?? null,
previousHash: entry.previousHash ?? null,
// Normalize timestamp to milliseconds since epoch to avoid precision issues.
timestamp: new Date(entry.timestamp).getTime()
};
private calculateHash(entry: any): string {
// Create a canonical object for hashing to ensure consistency in property order and types.
const objectToHash = {
actorIdentifier: entry.actorIdentifier,
actorIp: entry.actorIp ?? null,
actionType: entry.actionType,
targetType: entry.targetType ?? null,
targetId: entry.targetId ?? null,
details: entry.details ?? null,
previousHash: entry.previousHash ?? null,
// Normalize timestamp to milliseconds since epoch to avoid precision issues.
timestamp: new Date(entry.timestamp).getTime(),
};
const data = this.canonicalStringify(objectToHash);
return createHash('sha256').update(data).digest('hex');
}
const data = this.canonicalStringify(objectToHash);
return createHash('sha256').update(data).digest('hex');
}
private canonicalStringify(obj: any): string {
if (obj === undefined) {
return 'null';
}
if (obj === null || typeof obj !== 'object') {
return JSON.stringify(obj);
}
private canonicalStringify(obj: any): string {
if (obj === undefined) {
return 'null';
}
if (obj === null || typeof obj !== 'object') {
return JSON.stringify(obj);
}
if (Array.isArray(obj)) {
return `[${obj.map((item) => this.canonicalStringify(item)).join(',')}]`;
}
if (Array.isArray(obj)) {
return `[${obj.map((item) => this.canonicalStringify(item)).join(',')}]`;
}
const keys = Object.keys(obj).sort();
const pairs = keys.map((key) => {
const value = obj[key];
return `${JSON.stringify(key)}:${this.canonicalStringify(value)}`;
});
return `{${pairs.join(',')}}`;
}
const keys = Object.keys(obj).sort();
const pairs = keys.map((key) => {
const value = obj[key];
return `${JSON.stringify(key)}:${this.canonicalStringify(value)}`;
});
return `{${pairs.join(',')}}`;
}
public async getAuditLogs(options: GetAuditLogsOptions = {}): Promise<GetAuditLogsResponse> {
const {
page = 1,
limit = 20,
startDate,
endDate,
actor,
actionType,
sort = 'desc'
} = options;
public async getAuditLogs(options: GetAuditLogsOptions = {}): Promise<GetAuditLogsResponse> {
const {
page = 1,
limit = 20,
startDate,
endDate,
actor,
actionType,
sort = 'desc',
} = options;
const whereClauses = [];
if (startDate) whereClauses.push(gte(schema.auditLogs.timestamp, startDate));
if (endDate) whereClauses.push(lte(schema.auditLogs.timestamp, endDate));
if (actor) whereClauses.push(eq(schema.auditLogs.actorIdentifier, actor));
if (actionType) whereClauses.push(eq(schema.auditLogs.actionType, actionType));
const whereClauses = [];
if (startDate) whereClauses.push(gte(schema.auditLogs.timestamp, startDate));
if (endDate) whereClauses.push(lte(schema.auditLogs.timestamp, endDate));
if (actor) whereClauses.push(eq(schema.auditLogs.actorIdentifier, actor));
if (actionType) whereClauses.push(eq(schema.auditLogs.actionType, actionType));
const where = and(...whereClauses);
const where = and(...whereClauses);
const logs = await db.query.auditLogs.findMany({
where,
orderBy: [sort === 'asc' ? asc(schema.auditLogs.id) : desc(schema.auditLogs.id)],
limit,
offset: (page - 1) * limit
});
const logs = await db.query.auditLogs.findMany({
where,
orderBy: [sort === 'asc' ? asc(schema.auditLogs.id) : desc(schema.auditLogs.id)],
limit,
offset: (page - 1) * limit,
});
const totalResult = await db
.select({
count: sql<number>`count(*)`
})
.from(schema.auditLogs)
.where(where);
const totalResult = await db
.select({
count: sql<number>`count(*)`,
})
.from(schema.auditLogs)
.where(where);
const total = totalResult[0].count;
const total = totalResult[0].count;
return {
data: logs as AuditLogEntry[],
meta: {
total,
page,
limit
}
};
}
return {
data: logs as AuditLogEntry[],
meta: {
total,
page,
limit,
},
};
}
public async verifyAuditLog(): Promise<{ ok: boolean; message: string; logId?: number }> {
const chunkSize = 1000;
let offset = 0;
let previousHash: string | null = null;
/**
* TODO: create job for audit log verification, generate audit report (new DB table)
*/
while (true) {
const logs = await db.query.auditLogs.findMany({
orderBy: [asc(schema.auditLogs.id)],
limit: chunkSize,
offset
});
public async verifyAuditLog(): Promise<{ ok: boolean; message: string; logId?: number }> {
const chunkSize = 1000;
let offset = 0;
let previousHash: string | null = null;
/**
* TODO: create job for audit log verification, generate audit report (new DB table)
*/
while (true) {
const logs = await db.query.auditLogs.findMany({
orderBy: [asc(schema.auditLogs.id)],
limit: chunkSize,
offset,
});
if (logs.length === 0) {
break;
}
if (logs.length === 0) {
break;
}
for (const log of logs) {
if (log.previousHash !== previousHash) {
return {
ok: false,
message: 'Audit log chain is broken!',
logId: log.id
};
}
for (const log of logs) {
if (log.previousHash !== previousHash) {
return {
ok: false,
message: 'Audit log chain is broken!',
logId: log.id,
};
}
const calculatedHash = this.calculateHash(log);
const calculatedHash = this.calculateHash(log);
if (log.currentHash !== calculatedHash) {
return {
ok: false,
message: 'Audit log entry is tampered!',
logId: log.id
};
}
previousHash = log.currentHash;
}
if (log.currentHash !== calculatedHash) {
return {
ok: false,
message: 'Audit log entry is tampered!',
logId: log.id,
};
}
previousHash = log.currentHash;
}
offset += chunkSize;
}
offset += chunkSize;
}
return {
ok: true,
message: 'Audit log integrity verified successfully. The logs are not tempered with and the log chain is complete.'
};
}
return {
ok: true,
message:
'Audit log integrity verified successfully. The logs are not tempered with and the log chain is complete.',
};
}
}

6
packages/backend/src/services/AuthService.ts
View File

@@ -41,11 +41,7 @@ export class AuthService {
.sign(this.#jwtSecret);
}
public async login(
email: string,
password: string,
ip: string
): Promise<LoginResponse | null> {
public async login(email: string, password: string, ip: string): Promise<LoginResponse | null> {
const user = await this.#userService.findByEmail(email);
if (!user || !user.password) {

17
packages/backend/src/services/IndexingService.ts
View File

@@ -60,7 +60,6 @@ function sanitizeObject<T>(obj: T): T {
return obj;
}
export class IndexingService {
private dbService: DatabaseService;
private searchService: SearchService;
@@ -235,9 +234,7 @@ export class IndexingService {
/**
* @deprecated
*/
private async indexByEmail(
pendingEmail: PendingEmail
): Promise<void> {
private async indexByEmail(pendingEmail: PendingEmail): Promise<void> {
const attachments: AttachmentsType = [];
if (pendingEmail.email.attachments && pendingEmail.email.attachments.length > 0) {
for (const attachment of pendingEmail.email.attachments) {
@@ -259,7 +256,6 @@ export class IndexingService {
await this.searchService.addDocuments('emails', [document], 'id');
}
/**
* Creates a search document from a raw email object and its attachments.
*/
@@ -478,14 +474,12 @@ export class IndexingService {
'image/heif',
];
return extractableTypes.some((type) => mimeType.toLowerCase().includes(type));
}
/**
* Ensures all required fields are present in EmailDocument
*/
* Ensures all required fields are present in EmailDocument
*/
private ensureEmailDocumentFields(doc: Partial<EmailDocument>): EmailDocument {
return {
id: doc.id || 'missing-id',
@@ -510,7 +504,10 @@ export class IndexingService {
JSON.stringify(doc);
return true;
} catch (error) {
logger.error({ doc, error: (error as Error).message }, 'Invalid EmailDocument detected');
logger.error(
{ doc, error: (error as Error).message },
'Invalid EmailDocument detected'
);
return false;
}
}

2
packages/backend/src/services/IngestionService.ts
View File

@@ -186,7 +186,7 @@ export class IngestionService {
(key) =>
key !== 'providerConfig' &&
originalSource[key as keyof IngestionSource] !==
decryptedSource[key as keyof IngestionSource]
decryptedSource[key as keyof IngestionSource]
);
if (changedFields.length > 0) {
await this.auditService.createAuditLog({

146
packages/backend/src/services/IntegrityService.ts
View File

@@ -8,86 +8,86 @@ import type { IntegrityCheckResult } from '@open-archiver/types';
import { streamToBuffer } from '../helpers/streamToBuffer';
export class IntegrityService {
private storageService = new StorageService();
private storageService = new StorageService();
public async checkEmailIntegrity(emailId: string): Promise<IntegrityCheckResult[]> {
const results: IntegrityCheckResult[] = [];
public async checkEmailIntegrity(emailId: string): Promise<IntegrityCheckResult[]> {
const results: IntegrityCheckResult[] = [];
// 1. Fetch the archived email
const email = await db.query.archivedEmails.findFirst({
where: eq(archivedEmails.id, emailId),
});
// 1. Fetch the archived email
const email = await db.query.archivedEmails.findFirst({
where: eq(archivedEmails.id, emailId),
});
if (!email) {
throw new Error('Archived email not found');
}
if (!email) {
throw new Error('Archived email not found');
}
// 2. Check the email's integrity
const emailStream = await this.storageService.get(email.storagePath);
const emailBuffer = await streamToBuffer(emailStream);
const currentEmailHash = createHash('sha256').update(emailBuffer).digest('hex');
// 2. Check the email's integrity
const emailStream = await this.storageService.get(email.storagePath);
const emailBuffer = await streamToBuffer(emailStream);
const currentEmailHash = createHash('sha256').update(emailBuffer).digest('hex');
if (currentEmailHash === email.storageHashSha256) {
results.push({ type: 'email', id: email.id, isValid: true });
} else {
results.push({
type: 'email',
id: email.id,
isValid: false,
reason: 'Stored hash does not match current hash.',
});
}
if (currentEmailHash === email.storageHashSha256) {
results.push({ type: 'email', id: email.id, isValid: true });
} else {
results.push({
type: 'email',
id: email.id,
isValid: false,
reason: 'Stored hash does not match current hash.',
});
}
// 3. If the email has attachments, check them
if (email.hasAttachments) {
const emailAttachmentsRelations = await db.query.emailAttachments.findMany({
where: eq(emailAttachments.emailId, emailId),
with: {
attachment: true,
},
});
// 3. If the email has attachments, check them
if (email.hasAttachments) {
const emailAttachmentsRelations = await db.query.emailAttachments.findMany({
where: eq(emailAttachments.emailId, emailId),
with: {
attachment: true,
},
});
for (const relation of emailAttachmentsRelations) {
const attachment = relation.attachment;
try {
const attachmentStream = await this.storageService.get(attachment.storagePath);
const attachmentBuffer = await streamToBuffer(attachmentStream);
const currentAttachmentHash = createHash('sha256')
.update(attachmentBuffer)
.digest('hex');
for (const relation of emailAttachmentsRelations) {
const attachment = relation.attachment;
try {
const attachmentStream = await this.storageService.get(attachment.storagePath);
const attachmentBuffer = await streamToBuffer(attachmentStream);
const currentAttachmentHash = createHash('sha256')
.update(attachmentBuffer)
.digest('hex');
if (currentAttachmentHash === attachment.contentHashSha256) {
results.push({
type: 'attachment',
id: attachment.id,
filename: attachment.filename,
isValid: true,
});
} else {
results.push({
type: 'attachment',
id: attachment.id,
filename: attachment.filename,
isValid: false,
reason: 'Stored hash does not match current hash.',
});
}
} catch (error) {
logger.error(
{ attachmentId: attachment.id, error },
'Failed to read attachment from storage for integrity check.'
);
results.push({
type: 'attachment',
id: attachment.id,
filename: attachment.filename,
isValid: false,
reason: 'Could not read attachment file from storage.',
});
}
}
}
if (currentAttachmentHash === attachment.contentHashSha256) {
results.push({
type: 'attachment',
id: attachment.id,
filename: attachment.filename,
isValid: true,
});
} else {
results.push({
type: 'attachment',
id: attachment.id,
filename: attachment.filename,
isValid: false,
reason: 'Stored hash does not match current hash.',
});
}
} catch (error) {
logger.error(
{ attachmentId: attachment.id, error },
'Failed to read attachment from storage for integrity check.'
);
results.push({
type: 'attachment',
id: attachment.id,
filename: attachment.filename,
isValid: false,
reason: 'Could not read attachment file from storage.',
});
}
}
}
return results;
}
return results;
}
}

186
packages/backend/src/services/JobsService.ts
View File

@@ -1,107 +1,101 @@
import { Job, Queue } from 'bullmq';
import { ingestionQueue, indexingQueue } from '../jobs/queues';
import {
IJob,
IQueueCounts,
IQueueDetails,
IQueueOverview,
JobStatus,
} from '@open-archiver/types';
import { IJob, IQueueCounts, IQueueDetails, IQueueOverview, JobStatus } from '@open-archiver/types';
export class JobsService {
private queues: Queue[];
private queues: Queue[];
constructor() {
this.queues = [ingestionQueue, indexingQueue];
}
constructor() {
this.queues = [ingestionQueue, indexingQueue];
}
public async getQueues(): Promise<IQueueOverview[]> {
const queueOverviews: IQueueOverview[] = [];
for (const queue of this.queues) {
const counts = await queue.getJobCounts(
'active',
'completed',
'failed',
'delayed',
'waiting',
'paused'
);
queueOverviews.push({
name: queue.name,
counts: {
active: counts.active || 0,
completed: counts.completed || 0,
failed: counts.failed || 0,
delayed: counts.delayed || 0,
waiting: counts.waiting || 0,
paused: counts.paused || 0,
},
});
}
return queueOverviews;
}
public async getQueues(): Promise<IQueueOverview[]> {
const queueOverviews: IQueueOverview[] = [];
for (const queue of this.queues) {
const counts = await queue.getJobCounts(
'active',
'completed',
'failed',
'delayed',
'waiting',
'paused'
);
queueOverviews.push({
name: queue.name,
counts: {
active: counts.active || 0,
completed: counts.completed || 0,
failed: counts.failed || 0,
delayed: counts.delayed || 0,
waiting: counts.waiting || 0,
paused: counts.paused || 0,
},
});
}
return queueOverviews;
}
public async getQueueDetails(
queueName: string,
status: JobStatus,
page: number,
limit: number
): Promise<IQueueDetails> {
const queue = this.queues.find((q) => q.name === queueName);
if (!queue) {
throw new Error(`Queue ${queueName} not found`);
}
public async getQueueDetails(
queueName: string,
status: JobStatus,
page: number,
limit: number
): Promise<IQueueDetails> {
const queue = this.queues.find((q) => q.name === queueName);
if (!queue) {
throw new Error(`Queue ${queueName} not found`);
}
const counts = await queue.getJobCounts(
'active',
'completed',
'failed',
'delayed',
'waiting',
'paused'
);
const start = (page - 1) * limit;
const end = start + limit - 1;
const jobStatus = status === 'waiting' ? 'wait' : status;
const jobs = await queue.getJobs([jobStatus], start, end, true);
const totalJobs = await queue.getJobCountByTypes(jobStatus);
const counts = await queue.getJobCounts(
'active',
'completed',
'failed',
'delayed',
'waiting',
'paused'
);
const start = (page - 1) * limit;
const end = start + limit - 1;
const jobStatus = status === 'waiting' ? 'wait' : status;
const jobs = await queue.getJobs([jobStatus], start, end, true);
const totalJobs = await queue.getJobCountByTypes(jobStatus);
return {
name: queue.name,
counts: {
active: counts.active || 0,
completed: counts.completed || 0,
failed: counts.failed || 0,
delayed: counts.delayed || 0,
waiting: counts.waiting || 0,
paused: counts.paused || 0,
},
jobs: await Promise.all(jobs.map((job) => this.formatJob(job))),
pagination: {
currentPage: page,
totalPages: Math.ceil(totalJobs / limit),
totalJobs,
limit,
},
};
}
return {
name: queue.name,
counts: {
active: counts.active || 0,
completed: counts.completed || 0,
failed: counts.failed || 0,
delayed: counts.delayed || 0,
waiting: counts.waiting || 0,
paused: counts.paused || 0,
},
jobs: await Promise.all(jobs.map((job) => this.formatJob(job))),
pagination: {
currentPage: page,
totalPages: Math.ceil(totalJobs / limit),
totalJobs,
limit,
},
};
}
private async formatJob(job: Job): Promise<IJob> {
const state = await job.getState();
return {
id: job.id,
name: job.name,
data: job.data,
state: state,
failedReason: job.failedReason,
timestamp: job.timestamp,
processedOn: job.processedOn,
finishedOn: job.finishedOn,
attemptsMade: job.attemptsMade,
stacktrace: job.stacktrace,
returnValue: job.returnvalue,
ingestionSourceId: job.data.ingestionSourceId,
error: state === 'failed' ? job.stacktrace : undefined,
};
}
private async formatJob(job: Job): Promise<IJob> {
const state = await job.getState();
return {
id: job.id,
name: job.name,
data: job.data,
state: state,
failedReason: job.failedReason,
timestamp: job.timestamp,
processedOn: job.processedOn,
finishedOn: job.finishedOn,
attemptsMade: job.attemptsMade,
stacktrace: job.stacktrace,
returnValue: job.returnvalue,
ingestionSourceId: job.data.ingestionSourceId,
error: state === 'failed' ? job.stacktrace : undefined,
};
}
}

451
packages/backend/src/services/OcrService.ts
View File

@@ -3,269 +3,270 @@ import { logger } from '../config/logger';
// Simple LRU cache for Tika results with statistics
class TikaCache {
private cache = new Map<string, string>();
private maxSize = 50;
private hits = 0;
private misses = 0;
private cache = new Map<string, string>();
private maxSize = 50;
private hits = 0;
private misses = 0;
get(key: string): string | undefined {
const value = this.cache.get(key);
if (value !== undefined) {
this.hits++;
// LRU: Move element to the end
this.cache.delete(key);
this.cache.set(key, value);
} else {
this.misses++;
}
return value;
}
get(key: string): string | undefined {
const value = this.cache.get(key);
if (value !== undefined) {
this.hits++;
// LRU: Move element to the end
this.cache.delete(key);
this.cache.set(key, value);
} else {
this.misses++;
}
return value;
}
set(key: string, value: string): void {
// If already exists, delete first
if (this.cache.has(key)) {
this.cache.delete(key);
}
// If cache is full, remove oldest element
else if (this.cache.size >= this.maxSize) {
const firstKey = this.cache.keys().next().value;
if (firstKey !== undefined) {
this.cache.delete(firstKey);
}
}
set(key: string, value: string): void {
// If already exists, delete first
if (this.cache.has(key)) {
this.cache.delete(key);
}
// If cache is full, remove oldest element
else if (this.cache.size >= this.maxSize) {
const firstKey = this.cache.keys().next().value;
if (firstKey !== undefined) {
this.cache.delete(firstKey);
}
}
this.cache.set(key, value);
}
this.cache.set(key, value);
}
getStats(): { size: number; maxSize: number; hits: number; misses: number; hitRate: number } {
const total = this.hits + this.misses;
const hitRate = total > 0 ? (this.hits / total) * 100 : 0;
return {
size: this.cache.size,
maxSize: this.maxSize,
hits: this.hits,
misses: this.misses,
hitRate: Math.round(hitRate * 100) / 100 // 2 decimal places
};
}
getStats(): { size: number; maxSize: number; hits: number; misses: number; hitRate: number } {
const total = this.hits + this.misses;
const hitRate = total > 0 ? (this.hits / total) * 100 : 0;
return {
size: this.cache.size,
maxSize: this.maxSize,
hits: this.hits,
misses: this.misses,
hitRate: Math.round(hitRate * 100) / 100, // 2 decimal places
};
}
reset(): void {
this.cache.clear();
this.hits = 0;
this.misses = 0;
}
reset(): void {
this.cache.clear();
this.hits = 0;
this.misses = 0;
}
}
// Semaphore for running Tika requests
class TikaSemaphore {
private inProgress = new Map<string, Promise<string>>();
private waitCount = 0;
private inProgress = new Map<string, Promise<string>>();
private waitCount = 0;
async acquire(key: string, operation: () => Promise<string>): Promise<string> {
// Check if a request for this key is already running
const existingPromise = this.inProgress.get(key);
if (existingPromise) {
this.waitCount++;
logger.debug(`Waiting for in-progress Tika request (${key.slice(0, 8)}...)`);
try {
return await existingPromise;
} finally {
this.waitCount--;
}
}
async acquire(key: string, operation: () => Promise<string>): Promise<string> {
// Check if a request for this key is already running
const existingPromise = this.inProgress.get(key);
if (existingPromise) {
this.waitCount++;
logger.debug(`Waiting for in-progress Tika request (${key.slice(0, 8)}...)`);
try {
return await existingPromise;
} finally {
this.waitCount--;
}
}
// Start new request
const promise = this.executeOperation(key, operation);
this.inProgress.set(key, promise);
// Start new request
const promise = this.executeOperation(key, operation);
this.inProgress.set(key, promise);
try {
return await promise;
} finally {
// Remove promise from map when finished
this.inProgress.delete(key);
}
}
try {
return await promise;
} finally {
// Remove promise from map when finished
this.inProgress.delete(key);
}
}
private async executeOperation(key: string, operation: () => Promise<string>): Promise<string> {
try {
return await operation();
} catch (error) {
// Remove promise from map even on errors
logger.error(`Tika operation failed for key ${key.slice(0, 8)}...`, error);
throw error;
}
}
private async executeOperation(key: string, operation: () => Promise<string>): Promise<string> {
try {
return await operation();
} catch (error) {
// Remove promise from map even on errors
logger.error(`Tika operation failed for key ${key.slice(0, 8)}...`, error);
throw error;
}
}
getStats(): { inProgress: number; waitCount: number } {
return {
inProgress: this.inProgress.size,
waitCount: this.waitCount
};
}
getStats(): { inProgress: number; waitCount: number } {
return {
inProgress: this.inProgress.size,
waitCount: this.waitCount,
};
}
clear(): void {
this.inProgress.clear();
this.waitCount = 0;
}
clear(): void {
this.inProgress.clear();
this.waitCount = 0;
}
}
export class OcrService {
private tikaCache = new TikaCache();
private tikaSemaphore = new TikaSemaphore();
private tikaCache = new TikaCache();
private tikaSemaphore = new TikaSemaphore();
// Tika-based text extraction with cache and semaphore
async extractTextWithTika(buffer: Buffer, mimeType: string): Promise<string> {
const tikaUrl = process.env.TIKA_URL;
if (!tikaUrl) {
throw new Error('TIKA_URL environment variable not set');
}
// Tika-based text extraction with cache and semaphore
async extractTextWithTika(buffer: Buffer, mimeType: string): Promise<string> {
const tikaUrl = process.env.TIKA_URL;
if (!tikaUrl) {
throw new Error('TIKA_URL environment variable not set');
}
// Cache key: SHA-256 hash of the buffer
const hash = crypto.createHash('sha256').update(buffer).digest('hex');
// Cache key: SHA-256 hash of the buffer
const hash = crypto.createHash('sha256').update(buffer).digest('hex');
// Cache lookup (before semaphore!)
const cachedResult = this.tikaCache.get(hash);
if (cachedResult !== undefined) {
logger.debug(`Tika cache hit for ${mimeType} (${buffer.length} bytes)`);
return cachedResult;
}
// Cache lookup (before semaphore!)
const cachedResult = this.tikaCache.get(hash);
if (cachedResult !== undefined) {
logger.debug(`Tika cache hit for ${mimeType} (${buffer.length} bytes)`);
return cachedResult;
}
// Use semaphore to deduplicate parallel requests
return await this.tikaSemaphore.acquire(hash, async () => {
// Check cache again (might have been filled by parallel request)
const cachedAfterWait = this.tikaCache.get(hash);
if (cachedAfterWait !== undefined) {
logger.debug(`Tika cache hit after wait for ${mimeType} (${buffer.length} bytes)`);
return cachedAfterWait;
}
// Use semaphore to deduplicate parallel requests
return await this.tikaSemaphore.acquire(hash, async () => {
// Check cache again (might have been filled by parallel request)
const cachedAfterWait = this.tikaCache.get(hash);
if (cachedAfterWait !== undefined) {
logger.debug(`Tika cache hit after wait for ${mimeType} (${buffer.length} bytes)`);
return cachedAfterWait;
}
logger.debug(`Executing Tika request for ${mimeType} (${buffer.length} bytes)`);
logger.debug(`Executing Tika request for ${mimeType} (${buffer.length} bytes)`);
// DNS fallback: If "tika" hostname, also try localhost
const urlsToTry = [
`${tikaUrl}/tika`,
// Fallback falls DNS-Problem mit "tika" hostname
...(tikaUrl.includes('://tika:')
? [`${tikaUrl.replace('://tika:', '://localhost:')}/tika`]
: [])
];
// DNS fallback: If "tika" hostname, also try localhost
const urlsToTry = [
`${tikaUrl}/tika`,
// Fallback falls DNS-Problem mit "tika" hostname
...(tikaUrl.includes('://tika:')
? [`${tikaUrl.replace('://tika:', '://localhost:')}/tika`]
: []),
];
for (const url of urlsToTry) {
try {
logger.debug(`Trying Tika URL: ${url}`);
const response = await fetch(url, {
method: 'PUT',
headers: {
'Content-Type': mimeType || 'application/octet-stream',
Accept: 'text/plain',
Connection: 'close'
},
body: buffer,
signal: AbortSignal.timeout(180000)
});
for (const url of urlsToTry) {
try {
logger.debug(`Trying Tika URL: ${url}`);
const response = await fetch(url, {
method: 'PUT',
headers: {
'Content-Type': mimeType || 'application/octet-stream',
Accept: 'text/plain',
Connection: 'close',
},
body: buffer,
signal: AbortSignal.timeout(180000),
});
if (!response.ok) {
logger.warn(
`Tika extraction failed at ${url}: ${response.status} ${response.statusText}`
);
continue; // Try next URL
}
if (!response.ok) {
logger.warn(
`Tika extraction failed at ${url}: ${response.status} ${response.statusText}`
);
continue; // Try next URL
}
const text = await response.text();
const result = text.trim();
const text = await response.text();
const result = text.trim();
// Cache result (also empty strings to avoid repeated attempts)
this.tikaCache.set(hash, result);
// Cache result (also empty strings to avoid repeated attempts)
this.tikaCache.set(hash, result);
const cacheStats = this.tikaCache.getStats();
const semaphoreStats = this.tikaSemaphore.getStats();
logger.debug(
`Tika extraction successful - Cache: ${cacheStats.hits}H/${cacheStats.misses}M (${cacheStats.hitRate}%) - Semaphore: ${semaphoreStats.inProgress} active, ${semaphoreStats.waitCount} waiting`
);
const cacheStats = this.tikaCache.getStats();
const semaphoreStats = this.tikaSemaphore.getStats();
logger.debug(
`Tika extraction successful - Cache: ${cacheStats.hits}H/${cacheStats.misses}M (${cacheStats.hitRate}%) - Semaphore: ${semaphoreStats.inProgress} active, ${semaphoreStats.waitCount} waiting`
);
return result;
} catch (error) {
logger.warn(
`Tika extraction error at ${url}:`,
error instanceof Error ? error.message : 'Unknown error'
);
// Continue to next URL
}
}
return result;
} catch (error) {
logger.warn(
`Tika extraction error at ${url}:`,
error instanceof Error ? error.message : 'Unknown error'
);
// Continue to next URL
}
}
// All URLs failed - cache this too (as empty string)
logger.error('All Tika URLs failed');
this.tikaCache.set(hash, '');
return '';
});
}
// All URLs failed - cache this too (as empty string)
logger.error('All Tika URLs failed');
this.tikaCache.set(hash, '');
return '';
});
}
// Helper function to check Tika availability
async checkTikaAvailability(): Promise<boolean> {
const tikaUrl = process.env.TIKA_URL;
if (!tikaUrl) {
return false;
}
// Helper function to check Tika availability
async checkTikaAvailability(): Promise<boolean> {
const tikaUrl = process.env.TIKA_URL;
if (!tikaUrl) {
return false;
}
try {
const response = await fetch(`${tikaUrl}/version`, {
method: 'GET',
signal: AbortSignal.timeout(5000) // 5 seconds timeout
});
try {
const response = await fetch(`${tikaUrl}/version`, {
method: 'GET',
signal: AbortSignal.timeout(5000), // 5 seconds timeout
});
if (response.ok) {
const version = await response.text();
logger.info(`Tika server available, version: ${version.trim()}`);
return true;
}
if (response.ok) {
const version = await response.text();
logger.info(`Tika server available, version: ${version.trim()}`);
return true;
}
return false;
} catch (error) {
logger.warn(
'Tika server not available:',
error instanceof Error ? error.message : 'Unknown error'
);
return false;
}
}
return false;
} catch (error) {
logger.warn(
'Tika server not available:',
error instanceof Error ? error.message : 'Unknown error'
);
return false;
}
}
// Optional: Tika health check on startup
async initializeTextExtractor(): Promise<void> {
const tikaUrl = process.env.TIKA_URL;
// Optional: Tika health check on startup
async initializeTextExtractor(): Promise<void> {
const tikaUrl = process.env.TIKA_URL;
if (tikaUrl) {
const isAvailable = await this.checkTikaAvailability();
if (!isAvailable) {
logger.error(`Tika server configured but not available at: ${tikaUrl}`);
logger.error('Text extraction will fall back to legacy methods or fail');
}
} else {
logger.info('Using legacy text extraction methods (pdf2json, mammoth, xlsx)');
logger.info('Set TIKA_URL environment variable to use Apache Tika for better extraction');
}
}
if (tikaUrl) {
const isAvailable = await this.checkTikaAvailability();
if (!isAvailable) {
logger.error(`Tika server configured but not available at: ${tikaUrl}`);
logger.error('Text extraction will fall back to legacy methods or fail');
}
} else {
logger.info('Using legacy text extraction methods (pdf2json, mammoth, xlsx)');
logger.info(
'Set TIKA_URL environment variable to use Apache Tika for better extraction'
);
}
}
// Get cache statistics
getTikaCacheStats(): {
size: number;
maxSize: number;
hits: number;
misses: number;
hitRate: number;
} {
return this.tikaCache.getStats();
}
// Get cache statistics
getTikaCacheStats(): {
size: number;
maxSize: number;
hits: number;
misses: number;
hitRate: number;
} {
return this.tikaCache.getStats();
}
// Get semaphore statistics
getTikaSemaphoreStats(): { inProgress: number; waitCount: number } {
return this.tikaSemaphore.getStats();
}
// Get semaphore statistics
getTikaSemaphoreStats(): { inProgress: number; waitCount: number } {
return this.tikaSemaphore.getStats();
}
// Clear cache (e.g. for tests or manual reset)
clearTikaCache(): void {
this.tikaCache.reset();
this.tikaSemaphore.clear();
logger.info('Tika cache and semaphore cleared');
}
// Clear cache (e.g. for tests or manual reset)
clearTikaCache(): void {
this.tikaCache.reset();
this.tikaSemaphore.clear();
logger.info('Tika cache and semaphore cleared');
}
}

3
packages/backend/src/services/SettingsService.ts
View File

@@ -44,7 +44,8 @@ export class SettingsService {
const changedFields = Object.keys(newConfig).filter(
(key) =>
currentConfig[key as keyof SystemSettings] !== newConfig[key as keyof SystemSettings]
currentConfig[key as keyof SystemSettings] !==
newConfig[key as keyof SystemSettings]
);
if (changedFields.length > 0) {

4
packages/backend/src/services/StorageService.ts
View File

@@ -67,7 +67,9 @@ export class StorageService implements IStorageProvider {
async put(path: string, content: Buffer | NodeJS.ReadableStream): Promise<void> {
const buffer =
content instanceof Buffer ? content : await streamToBuffer(content as NodeJS.ReadableStream);
content instanceof Buffer
? content
: await streamToBuffer(content as NodeJS.ReadableStream);
const encryptedContent = await this.encrypt(buffer);
return this.provider.put(path, encryptedContent);
}

12
packages/backend/src/services/ingestion-connectors/ImapConnector.ts
View File

@@ -8,7 +8,7 @@ import type {
import type { IEmailConnector } from '../EmailProviderFactory';
import { ImapFlow } from 'imapflow';
import { simpleParser, ParsedMail, Attachment, AddressObject, Headers } from 'mailparser';
import { config } from '../../config'
import { config } from '../../config';
import { logger } from '../../config/logger';
import { getThreadId } from './helpers/utils';
@@ -161,18 +161,12 @@ export class ImapConnector implements IEmailConnector {
// filter out junk/spam mail emails
if (mailbox.specialUse) {
const specialUse = mailbox.specialUse.toLowerCase();
if (
specialUse === '\\junk' ||
specialUse === '\\trash'
) {
if (specialUse === '\\junk' || specialUse === '\\trash') {
return false;
}
}
// Fallback to checking flags
if (
mailbox.flags.has('\\Trash') ||
mailbox.flags.has('\\Junk')
) {
if (mailbox.flags.has('\\Trash') || mailbox.flags.has('\\Junk')) {
return false;
}

286
packages/backend/src/services/ingestion-connectors/MboxConnector.ts
View File

@@ -1,9 +1,9 @@
import type {
MboxImportCredentials,
EmailObject,
EmailAddress,
SyncState,
MailboxUser,
MboxImportCredentials,
EmailObject,
EmailAddress,
SyncState,
MailboxUser,
} from '@open-archiver/types';
import type { IEmailConnector } from '../EmailProviderFactory';
import { simpleParser, ParsedMail, Attachment, AddressObject } from 'mailparser';
@@ -15,160 +15,160 @@ import { createHash } from 'crypto';
import { streamToBuffer } from '../../helpers/streamToBuffer';
export class MboxConnector implements IEmailConnector {
private storage: StorageService;
private storage: StorageService;
constructor(private credentials: MboxImportCredentials) {
this.storage = new StorageService();
}
constructor(private credentials: MboxImportCredentials) {
this.storage = new StorageService();
}
public async testConnection(): Promise<boolean> {
try {
if (!this.credentials.uploadedFilePath) {
throw Error('Mbox file path not provided.');
}
if (!this.credentials.uploadedFilePath.includes('.mbox')) {
throw Error('Provided file is not in the MBOX format.');
}
const fileExist = await this.storage.exists(this.credentials.uploadedFilePath);
if (!fileExist) {
throw Error('Mbox file upload not finished yet, please wait.');
}
public async testConnection(): Promise<boolean> {
try {
if (!this.credentials.uploadedFilePath) {
throw Error('Mbox file path not provided.');
}
if (!this.credentials.uploadedFilePath.includes('.mbox')) {
throw Error('Provided file is not in the MBOX format.');
}
const fileExist = await this.storage.exists(this.credentials.uploadedFilePath);
if (!fileExist) {
throw Error('Mbox file upload not finished yet, please wait.');
}
return true;
} catch (error) {
logger.error({ error, credentials: this.credentials }, 'Mbox file validation failed.');
throw error;
}
}
return true;
} catch (error) {
logger.error({ error, credentials: this.credentials }, 'Mbox file validation failed.');
throw error;
}
}
public async *listAllUsers(): AsyncGenerator<MailboxUser> {
const displayName =
this.credentials.uploadedFileName || `mbox-import-${new Date().getTime()}`;
logger.info(`Found potential mailbox: ${displayName}`);
const constructedPrimaryEmail = `${displayName.replace(/ /g, '.').toLowerCase()}@mbox.local`;
yield {
id: constructedPrimaryEmail,
primaryEmail: constructedPrimaryEmail,
displayName: displayName,
};
}
public async *listAllUsers(): AsyncGenerator<MailboxUser> {
const displayName =
this.credentials.uploadedFileName || `mbox-import-${new Date().getTime()}`;
logger.info(`Found potential mailbox: ${displayName}`);
const constructedPrimaryEmail = `${displayName.replace(/ /g, '.').toLowerCase()}@mbox.local`;
yield {
id: constructedPrimaryEmail,
primaryEmail: constructedPrimaryEmail,
displayName: displayName,
};
}
public async *fetchEmails(
userEmail: string,
syncState?: SyncState | null
): AsyncGenerator<EmailObject | null> {
try {
const fileStream = await this.storage.get(this.credentials.uploadedFilePath);
const fileBuffer = await streamToBuffer(fileStream as Readable);
const mboxContent = fileBuffer.toString('utf-8');
const emailDelimiter = '\nFrom ';
const emails = mboxContent.split(emailDelimiter);
public async *fetchEmails(
userEmail: string,
syncState?: SyncState | null
): AsyncGenerator<EmailObject | null> {
try {
const fileStream = await this.storage.get(this.credentials.uploadedFilePath);
const fileBuffer = await streamToBuffer(fileStream as Readable);
const mboxContent = fileBuffer.toString('utf-8');
const emailDelimiter = '\nFrom ';
const emails = mboxContent.split(emailDelimiter);
// The first split part might be empty or part of the first email's header, so we adjust.
if (emails.length > 0 && !mboxContent.startsWith('From ')) {
emails.shift(); // Adjust if the file doesn't start with "From "
}
// The first split part might be empty or part of the first email's header, so we adjust.
if (emails.length > 0 && !mboxContent.startsWith('From ')) {
emails.shift(); // Adjust if the file doesn't start with "From "
}
logger.info(`Found ${emails.length} potential emails in the mbox file.`);
let emailCount = 0;
logger.info(`Found ${emails.length} potential emails in the mbox file.`);
let emailCount = 0;
for (const email of emails) {
try {
// Re-add the "From " delimiter for the parser, except for the very first email
const emailWithDelimiter =
emailCount > 0 || mboxContent.startsWith('From ') ? `From ${email}` : email;
const emailBuffer = Buffer.from(emailWithDelimiter, 'utf-8');
const emailObject = await this.parseMessage(emailBuffer, '');
yield emailObject;
emailCount++;
} catch (error) {
logger.error(
{ error, file: this.credentials.uploadedFilePath },
'Failed to process a single message from mbox file. Skipping.'
);
}
}
logger.info(`Finished processing mbox file. Total emails processed: ${emailCount}`);
} finally {
try {
await this.storage.delete(this.credentials.uploadedFilePath);
} catch (error) {
logger.error(
{ error, file: this.credentials.uploadedFilePath },
'Failed to delete mbox file after processing.'
);
}
}
}
for (const email of emails) {
try {
// Re-add the "From " delimiter for the parser, except for the very first email
const emailWithDelimiter =
emailCount > 0 || mboxContent.startsWith('From ') ? `From ${email}` : email;
const emailBuffer = Buffer.from(emailWithDelimiter, 'utf-8');
const emailObject = await this.parseMessage(emailBuffer, '');
yield emailObject;
emailCount++;
} catch (error) {
logger.error(
{ error, file: this.credentials.uploadedFilePath },
'Failed to process a single message from mbox file. Skipping.'
);
}
}
logger.info(`Finished processing mbox file. Total emails processed: ${emailCount}`);
} finally {
try {
await this.storage.delete(this.credentials.uploadedFilePath);
} catch (error) {
logger.error(
{ error, file: this.credentials.uploadedFilePath },
'Failed to delete mbox file after processing.'
);
}
}
}
private async parseMessage(emlBuffer: Buffer, path: string): Promise<EmailObject> {
const parsedEmail: ParsedMail = await simpleParser(emlBuffer);
private async parseMessage(emlBuffer: Buffer, path: string): Promise<EmailObject> {
const parsedEmail: ParsedMail = await simpleParser(emlBuffer);
const attachments = parsedEmail.attachments.map((attachment: Attachment) => ({
filename: attachment.filename || 'untitled',
contentType: attachment.contentType,
size: attachment.size,
content: attachment.content as Buffer,
}));
const attachments = parsedEmail.attachments.map((attachment: Attachment) => ({
filename: attachment.filename || 'untitled',
contentType: attachment.contentType,
size: attachment.size,
content: attachment.content as Buffer,
}));
const mapAddresses = (
addresses: AddressObject | AddressObject[] | undefined
): EmailAddress[] => {
if (!addresses) return [];
const addressArray = Array.isArray(addresses) ? addresses : [addresses];
return addressArray.flatMap((a) =>
a.value.map((v) => ({
name: v.name,
address: v.address?.replaceAll(`'`, '') || '',
}))
);
};
const mapAddresses = (
addresses: AddressObject | AddressObject[] | undefined
): EmailAddress[] => {
if (!addresses) return [];
const addressArray = Array.isArray(addresses) ? addresses : [addresses];
return addressArray.flatMap((a) =>
a.value.map((v) => ({
name: v.name,
address: v.address?.replaceAll(`'`, '') || '',
}))
);
};
const threadId = getThreadId(parsedEmail.headers);
let messageId = parsedEmail.messageId;
const threadId = getThreadId(parsedEmail.headers);
let messageId = parsedEmail.messageId;
if (!messageId) {
messageId = `generated-${createHash('sha256').update(emlBuffer).digest('hex')}`;
}
if (!messageId) {
messageId = `generated-${createHash('sha256').update(emlBuffer).digest('hex')}`;
}
const from = mapAddresses(parsedEmail.from);
if (from.length === 0) {
from.push({ name: 'No Sender', address: 'No Sender' });
}
const from = mapAddresses(parsedEmail.from);
if (from.length === 0) {
from.push({ name: 'No Sender', address: 'No Sender' });
}
// Extract folder path from headers. Mbox files don't have a standard folder structure, so we rely on custom headers added by email clients.
// Gmail uses 'X-Gmail-Labels', and other clients like Thunderbird may use 'X-Folder'.
const gmailLabels = parsedEmail.headers.get('x-gmail-labels');
const folderHeader = parsedEmail.headers.get('x-folder');
let finalPath = '';
// Extract folder path from headers. Mbox files don't have a standard folder structure, so we rely on custom headers added by email clients.
// Gmail uses 'X-Gmail-Labels', and other clients like Thunderbird may use 'X-Folder'.
const gmailLabels = parsedEmail.headers.get('x-gmail-labels');
const folderHeader = parsedEmail.headers.get('x-folder');
let finalPath = '';
if (gmailLabels && typeof gmailLabels === 'string') {
// We take the first label as the primary folder.
// Gmail labels can be hierarchical, but we'll simplify to the first label.
finalPath = gmailLabels.split(',')[0];
} else if (folderHeader && typeof folderHeader === 'string') {
finalPath = folderHeader;
}
if (gmailLabels && typeof gmailLabels === 'string') {
// We take the first label as the primary folder.
// Gmail labels can be hierarchical, but we'll simplify to the first label.
finalPath = gmailLabels.split(',')[0];
} else if (folderHeader && typeof folderHeader === 'string') {
finalPath = folderHeader;
}
return {
id: messageId,
threadId: threadId,
from,
to: mapAddresses(parsedEmail.to),
cc: mapAddresses(parsedEmail.cc),
bcc: mapAddresses(parsedEmail.bcc),
subject: parsedEmail.subject || '',
body: parsedEmail.text || '',
html: parsedEmail.html || '',
headers: parsedEmail.headers,
attachments,
receivedAt: parsedEmail.date || new Date(),
eml: emlBuffer,
path: finalPath,
};
}
return {
id: messageId,
threadId: threadId,
from,
to: mapAddresses(parsedEmail.to),
cc: mapAddresses(parsedEmail.cc),
bcc: mapAddresses(parsedEmail.bcc),
subject: parsedEmail.subject || '',
body: parsedEmail.text || '',
html: parsedEmail.html || '',
headers: parsedEmail.headers,
attachments,
receivedAt: parsedEmail.date || new Date(),
eml: emlBuffer,
path: finalPath,
};
}
public getUpdatedSyncState(): SyncState {
return {};
}
public getUpdatedSyncState(): SyncState {
return {};
}
}

4
packages/backend/src/services/ingestion-connectors/PSTConnector.ts
View File

@@ -281,8 +281,8 @@ export class PSTConnector implements IEmailConnector {
emlBuffer ?? Buffer.from(parsedEmail.text || parsedEmail.html || '', 'utf-8')
)
.digest('hex')}-${createHash('sha256')
.update(emlBuffer ?? Buffer.from(msg.subject || '', 'utf-8'))
.digest('hex')}-${msg.clientSubmitTime?.getTime()}`;
.update(emlBuffer ?? Buffer.from(msg.subject || '', 'utf-8'))
.digest('hex')}-${msg.clientSubmitTime?.getTime()}`;
}
return {
id: messageId,

6
packages/enterprise/src/index.ts
View File

@@ -4,7 +4,7 @@ import { auditLogModule } from './modules/audit-log/audit-log.module';
import { licenseModule } from './modules/license/license.module';
export const enterpriseModules: ArchiverModule[] = [
licenseModule,
retentionPolicyModule,
auditLogModule
licenseModule,
retentionPolicyModule,
auditLogModule,
];

35
packages/enterprise/src/middleware/featureEnabled.ts
View File

@@ -11,21 +11,24 @@ import { logger } from '@open-archiver/backend';
* @returns An Express middleware function.
*/
export const featureEnabled = (feature: OpenArchiverFeature) => {
return (req: Request, res: Response, next: NextFunction) => {
try {
if (licenseService.isFeatureEnabled(feature)) {
return next();
}
return (req: Request, res: Response, next: NextFunction) => {
try {
if (licenseService.isFeatureEnabled(feature)) {
return next();
}
res.status(403).json({
error: 'Forbidden',
message: `This feature (${feature}) is not enabled for your current license. Please upgrade your plan to access this feature.`,
});
} catch (error) {
// In case of an unexpected error during license verification,
// log the error but allow the request to proceed.
logger.error(`🚨 CRITICAL: License check failed for feature "${feature}". Allowing access by default. Error:`, error);
return next();
}
};
res.status(403).json({
error: 'Forbidden',
message: `This feature (${feature}) is not enabled for your current license. Please upgrade your plan to access this feature.`,
});
} catch (error) {
// In case of an unexpected error during license verification,
// log the error but allow the request to proceed.
logger.error(
`🚨 CRITICAL: License check failed for feature "${feature}". Allowing access by default. Error:`,
error
);
return next();
}
};
};

60
packages/enterprise/src/modules/audit-log/audit-log.controller.ts
View File

@@ -4,38 +4,40 @@ import { AuditLogActions, AuditLogTargetTypes } from '@open-archiver/types';
import { z } from 'zod';
const getAuditLogsSchema = z.object({
page: z.coerce.number().min(1).optional(),
limit: z.coerce.number().min(1).max(100).optional(),
startDate: z.coerce.date().optional(),
endDate: z.coerce.date().optional(),
actor: z.string().optional(),
action: z.enum(AuditLogActions).optional(),
targetType: z.enum(AuditLogTargetTypes).optional(),
sort: z.enum(['asc', 'desc']).optional()
page: z.coerce.number().min(1).optional(),
limit: z.coerce.number().min(1).max(100).optional(),
startDate: z.coerce.date().optional(),
endDate: z.coerce.date().optional(),
actor: z.string().optional(),
action: z.enum(AuditLogActions).optional(),
targetType: z.enum(AuditLogTargetTypes).optional(),
sort: z.enum(['asc', 'desc']).optional(),
});
export class AuditLogController {
private auditService = new AuditService();
private auditService = new AuditService();
public getAuditLogs = async (req: Request, res: Response) => {
try {
const query = getAuditLogsSchema.parse(req.query);
const result = await this.auditService.getAuditLogs(query);
res.status(200).json(result);
} catch (error) {
if (error instanceof z.ZodError) {
return res.status(400).json({ message: 'Invalid query parameters', errors: error.issues });
}
res.status(500).json({ message: 'Internal server error.' });
}
};
public getAuditLogs = async (req: Request, res: Response) => {
try {
const query = getAuditLogsSchema.parse(req.query);
const result = await this.auditService.getAuditLogs(query);
res.status(200).json(result);
} catch (error) {
if (error instanceof z.ZodError) {
return res
.status(400)
.json({ message: 'Invalid query parameters', errors: error.issues });
}
res.status(500).json({ message: 'Internal server error.' });
}
};
public verifyAuditLog = async (req: Request, res: Response) => {
const result = await this.auditService.verifyAuditLog();
if (result.ok) {
res.status(200).json(result);
} else {
res.status(500).json(result);
}
};
public verifyAuditLog = async (req: Request, res: Response) => {
const result = await this.auditService.verifyAuditLog();
if (result.ok) {
res.status(200).json(result);
} else {
res.status(500).json(result);
}
};
}

8
packages/enterprise/src/modules/audit-log/audit-log.module.ts
View File

@@ -6,11 +6,11 @@ import { config } from '@open-archiver/backend';
import { OpenArchiverFeature } from '@open-archiver/types';
class AuditLogModule implements ArchiverModule {
name: OpenArchiverFeature = OpenArchiverFeature.AUDIT_LOG;
name: OpenArchiverFeature = OpenArchiverFeature.AUDIT_LOG;
async initialize(app: Express, authService: AuthService): Promise<void> {
app.use(`/${config.api.version}/enterprise/audit-logs`, auditLogRoutes(authService));
}
async initialize(app: Express, authService: AuthService): Promise<void> {
app.use(`/${config.api.version}/enterprise/audit-logs`, auditLogRoutes(authService));
}
}
export const auditLogModule = new AuditLogModule();

18
packages/enterprise/src/modules/audit-log/audit-log.routes.ts
View File

@@ -5,20 +5,14 @@ import { AuthService } from '@open-archiver/backend';
import { featureEnabled } from '../../middleware/featureEnabled';
import { OpenArchiverFeature } from '@open-archiver/types';
export const auditLogRoutes = (authService: AuthService): Router => {
const router = Router();
const controller = new AuditLogController();
router.use(requireAuth(authService), featureEnabled(OpenArchiverFeature.AUDIT_LOG));
const router = Router();
const controller = new AuditLogController();
router.use(requireAuth(authService), featureEnabled(OpenArchiverFeature.AUDIT_LOG));
router.get('/',
requirePermission('manage', 'all'),
controller.getAuditLogs);
router.get('/', requirePermission('manage', 'all'), controller.getAuditLogs);
router.post('/verify',
requirePermission('manage', 'all'),
controller.verifyAuditLog);
router.post('/verify', requirePermission('manage', 'all'), controller.verifyAuditLog);
return router;
return router;
};

110
packages/enterprise/src/modules/license/LicenseReportingService.ts
View File

@@ -2,80 +2,76 @@ import * as cron from 'node-cron';
import * as fs from 'fs/promises';
import * as path from 'path';
import { LicenseFilePayload, LicenseStatusPayload } from '@open-archiver/types';
import {
logger,
db,
drizzleOrm,
archivedEmails
} from '@open-archiver/backend';
import { logger, db, drizzleOrm, archivedEmails } from '@open-archiver/backend';
// license server is yet to be implemented.
const LICENSE_SERVER_URL = 'https://licensing.openarchiver.com/api/v1/ping';
export const CACHE_FILE_PATH = path.join(__dirname, 'license-status.json');
class LicenseReportingService {
private licensePayload: LicenseFilePayload | null = null;
private licensePayload: LicenseFilePayload | null = null;
public start(payload: LicenseFilePayload) {
this.licensePayload = payload;
// Schedule to run once every 24 hours, with a random minute/hour to distribute load.
const cronExpression = `${Math.floor(Math.random() * 60)} ${Math.floor(Math.random() * 5)} * * *`;
public start(payload: LicenseFilePayload) {
this.licensePayload = payload;
// Schedule to run once every 24 hours, with a random minute/hour to distribute load.
const cronExpression = `${Math.floor(Math.random() * 60)} ${Math.floor(Math.random() * 5)} * * *`;
cron.schedule(cronExpression, () => {
this.phoneHome();
});
cron.schedule(cronExpression, () => {
this.phoneHome();
});
logger.info(`📞 License reporting service scheduled with expression: ${cronExpression}`);
}
logger.info(`📞 License reporting service scheduled with expression: ${cronExpression}`);
}
public async phoneHome() {
if (!this.licensePayload) {
logger.warn('📞 Phone home skipped: License payload not loaded.');
return;
}
public async phoneHome() {
if (!this.licensePayload) {
logger.warn('📞 Phone home skipped: License payload not loaded.');
return;
}
try {
// 1. Count Active Seats, the result will be used to send to license server.
const activeSeats = await this.countActiveSeats();
try {
// 1. Count Active Seats, the result will be used to send to license server.
const activeSeats = await this.countActiveSeats();
logger.info(`Performing daily license check for ${this.licensePayload.customerName}. Active seats: ${activeSeats}`);
logger.info(
`Performing daily license check for ${this.licensePayload.customerName}. Active seats: ${activeSeats}`
);
// 2. Phone Home (mocked for now)
// will be replaced by a fetch call to the license server.
const mockedResponse: LicenseStatusPayload = {
status: 'VALID'
};
// 2. Phone Home (mocked for now)
// will be replaced by a fetch call to the license server.
const mockedResponse: LicenseStatusPayload = {
status: 'VALID',
};
// 3. Cache Response
await this.cacheLicenseStatus(mockedResponse);
// 3. Cache Response
await this.cacheLicenseStatus(mockedResponse);
} catch (error) {
logger.error('Phone home failed:', error);
// If the request fails, we do nothing and the app continues with the last known status.
}
}
} catch (error) {
logger.error('Phone home failed:', error);
// If the request fails, we do nothing and the app continues with the last known status.
}
}
public async countActiveSeats(): Promise<number> {
try {
const result = await db
.select({ count: drizzleOrm.countDistinct(archivedEmails.userEmail) })
.from(archivedEmails);
public async countActiveSeats(): Promise<number> {
try {
const result = await db
.select({ count: drizzleOrm.countDistinct(archivedEmails.userEmail) })
.from(archivedEmails);
return result[0]?.count || 0;
} catch (error) {
logger.error('Failed to count active seats from database:', error);
return 0; // Return 0 if the query fails to avoid breaking the process.
}
}
return result[0]?.count || 0;
} catch (error) {
logger.error('Failed to count active seats from database:', error);
return 0; // Return 0 if the query fails to avoid breaking the process.
}
}
private async cacheLicenseStatus(status: LicenseStatusPayload) {
try {
await fs.writeFile(CACHE_FILE_PATH, JSON.stringify(status, null, 2));
logger.info(`License status successfully cached to ${CACHE_FILE_PATH}`);
} catch (error) {
logger.error(`Failed to cache license status to ${CACHE_FILE_PATH}:`, error);
}
}
private async cacheLicenseStatus(status: LicenseStatusPayload) {
try {
await fs.writeFile(CACHE_FILE_PATH, JSON.stringify(status, null, 2));
logger.info(`License status successfully cached to ${CACHE_FILE_PATH}`);
} catch (error) {
logger.error(`Failed to cache license status to ${CACHE_FILE_PATH}:`, error);
}
}
}
export const licenseReportingService = new LicenseReportingService();

172
packages/enterprise/src/modules/license/LicenseService.ts
View File

@@ -1,7 +1,11 @@
import * as fs from 'fs';
import * as path from 'path';
import * as jwt from 'jsonwebtoken';
import { LicenseFilePayload, LicenseStatusPayload, OpenArchiverFeature } from '@open-archiver/types';
import {
LicenseFilePayload,
LicenseStatusPayload,
OpenArchiverFeature,
} from '@open-archiver/types';
import { logger } from '@open-archiver/backend';
import { licenseReportingService } from './LicenseReportingService';
import { CACHE_FILE_PATH } from './LicenseReportingService';
@@ -16,98 +20,102 @@ X0SrmkMxGyEd18vMXBGD9piAR3MTRskQ6XOoEo0fio6s9LtgPzKkJBdyVg==
type LicenseStatus = 'VALID' | 'INVALID' | 'EXPIRED' | 'NOT_FOUND';
class LicenseService {
public licensePayload: LicenseFilePayload | null = null;
public licenseStatus: LicenseStatus = 'NOT_FOUND';
public cachedStatus: LicenseStatusPayload | null = null;
public licensePayload: LicenseFilePayload | null = null;
public licenseStatus: LicenseStatus = 'NOT_FOUND';
public cachedStatus: LicenseStatusPayload | null = null;
constructor() {
this.loadAndVerifyLicense();
this.loadCachedStatus();
}
constructor() {
this.loadAndVerifyLicense();
this.loadCachedStatus();
}
private loadAndVerifyLicense() {
try {
const licenseKey = process.env.OA_LICENSE_KEY || fs.readFileSync(path.join(__dirname, 'license.jwt'), 'utf-8');
private loadAndVerifyLicense() {
try {
const licenseKey =
process.env.OA_LICENSE_KEY ||
fs.readFileSync(path.join(__dirname, 'license.jwt'), 'utf-8');
if (!licenseKey) {
this.licenseStatus = 'NOT_FOUND';
logger.warn('📄 License key not found.');
return;
}
if (!licenseKey) {
this.licenseStatus = 'NOT_FOUND';
logger.warn('📄 License key not found.');
return;
}
const decoded = jwt.verify(licenseKey, PUBLIC_KEY, {
algorithms: ['ES256'],
}) as LicenseFilePayload;
const decoded = jwt.verify(licenseKey, PUBLIC_KEY, {
algorithms: ['ES256'],
}) as LicenseFilePayload;
this.licensePayload = decoded;
this.licenseStatus = 'VALID';
logger.info(`Enterprise license successfully verified for: ${this.licensePayload.customerName}`);
// Start the reporting service now that we have a valid license payload
licenseReportingService.start(this.licensePayload);
this.licensePayload = decoded;
this.licenseStatus = 'VALID';
logger.info(
`Enterprise license successfully verified for: ${this.licensePayload.customerName}`
);
// Start the reporting service now that we have a valid license payload
licenseReportingService.start(this.licensePayload);
} catch (error) {
if (error instanceof jwt.TokenExpiredError) {
this.licenseStatus = 'EXPIRED';
logger.error('License verification failed: The license has expired.');
} else if (error instanceof jwt.JsonWebTokenError) {
this.licenseStatus = 'INVALID';
logger.error(`License verification failed: ${error.message}`);
} else {
this.licenseStatus = 'INVALID';
logger.error('An unexpected error occurred during license verification:', error);
}
}
}
} catch (error) {
if (error instanceof jwt.TokenExpiredError) {
this.licenseStatus = 'EXPIRED';
logger.error('License verification failed: The license has expired.');
} else if (error instanceof jwt.JsonWebTokenError) {
this.licenseStatus = 'INVALID';
logger.error(`License verification failed: ${error.message}`);
} else {
this.licenseStatus = 'INVALID';
logger.error('An unexpected error occurred during license verification:', error);
}
}
}
private async loadCachedStatus() {
try {
if (fs.existsSync(CACHE_FILE_PATH)) {
const data = fs.readFileSync(CACHE_FILE_PATH, 'utf-8');
this.cachedStatus = JSON.parse(data);
logger.info(
`Successfully loaded cached license status: ${this.cachedStatus?.status}`
);
} else {
// On a new installation, the cache file won't exist. We default to a valid state
logger.info(
`License status cache not found. Assuming 'VALID' until first phone-home.`
);
this.cachedStatus = { status: 'VALID' };
}
} catch (error) {
logger.error(`Failed to load or initialize cached license status:`, error);
// Fallback to a valid status if parsing fails to prevent locking out users.
this.cachedStatus = { status: 'VALID' };
}
}
private async loadCachedStatus() {
try {
if (fs.existsSync(CACHE_FILE_PATH)) {
const data = fs.readFileSync(CACHE_FILE_PATH, 'utf-8');
this.cachedStatus = JSON.parse(data);
logger.info(`Successfully loaded cached license status: ${this.cachedStatus?.status}`);
} else {
// On a new installation, the cache file won't exist. We default to a valid state
logger.info(`License status cache not found. Assuming 'VALID' until first phone-home.`);
this.cachedStatus = { status: 'VALID' };
}
} catch (error) {
logger.error(`Failed to load or initialize cached license status:`, error);
// Fallback to a valid status if parsing fails to prevent locking out users.
this.cachedStatus = { status: 'VALID' };
}
}
public isFeatureEnabled(feature: OpenArchiverFeature): boolean {
// A license payload must exist to know which features are granted.
if (!this.licensePayload) {
return false;
}
public isFeatureEnabled(feature: OpenArchiverFeature): boolean {
// A license payload must exist to know which features are granted.
if (!this.licensePayload) {
return false;
}
// Check if the license is supposed to grant the feature.
const hasAllFeatures = this.licensePayload.features.includes(OpenArchiverFeature.ALL);
const hasSpecificFeature = this.licensePayload.features.includes(feature);
// Check if the license is supposed to grant the feature.
const hasAllFeatures = this.licensePayload.features.includes(OpenArchiverFeature.ALL);
const hasSpecificFeature = this.licensePayload.features.includes(feature);
if (!hasAllFeatures && !hasSpecificFeature) {
return false;
}
if (!hasAllFeatures && !hasSpecificFeature) {
return false;
}
// Now, check the validity. The server's cached status is the highest source of truth.
if (this.cachedStatus?.status === 'REVOKED') {
if (this.cachedStatus.gracePeriodEnds) {
const gracePeriodEnd = new Date(this.cachedStatus.gracePeriodEnds);
// The grace period is active, so the feature is enabled regardless of local JWT status.
return new Date() < gracePeriodEnd;
}
// Revoked and no grace period.
return false;
}
// Now, check the validity. The server's cached status is the highest source of truth.
if (this.cachedStatus?.status === 'REVOKED') {
if (this.cachedStatus.gracePeriodEnds) {
const gracePeriodEnd = new Date(this.cachedStatus.gracePeriodEnds);
// The grace period is active, so the feature is enabled regardless of local JWT status.
return new Date() < gracePeriodEnd;
}
// Revoked and no grace period.
return false;
}
// If not revoked by the server, the local license JWT must be valid.
return this.licenseStatus === 'VALID';
}
// If not revoked by the server, the local license JWT must be valid.
return this.licenseStatus === 'VALID';
}
}
export const licenseService = new LicenseService();

47
packages/enterprise/src/modules/license/license.controller.ts
View File

@@ -4,33 +4,36 @@ import { licenseService } from './LicenseService';
import { licenseReportingService } from './LicenseReportingService';
class LicenseController {
public getLicenseStatus = async (req: Request, res: Response) => {
if (!licenseService.licensePayload) {
return res.status(404).json({ error: 'License information not found.' });
}
public getLicenseStatus = async (req: Request, res: Response) => {
if (!licenseService.licensePayload) {
return res.status(404).json({ error: 'License information not found.' });
}
const activeSeats = await licenseReportingService.countActiveSeats()
const activeSeats = await licenseReportingService.countActiveSeats();
const allPossibleFeatures: OpenArchiverFeature[] = Object.values(OpenArchiverFeature);
const allPossibleFeatures: OpenArchiverFeature[] = Object.values(OpenArchiverFeature);
const features = allPossibleFeatures.reduce((acc, feature) => {
acc[feature] = licenseService.isFeatureEnabled(feature);
return acc;
}, {} as { [key in OpenArchiverFeature]?: boolean });
const features = allPossibleFeatures.reduce(
(acc, feature) => {
acc[feature] = licenseService.isFeatureEnabled(feature);
return acc;
},
{} as { [key in OpenArchiverFeature]?: boolean }
);
const response: ConsolidatedLicenseStatus = {
customerName: licenseService.licensePayload.customerName,
planSeats: licenseService.licensePayload.planSeats,
expiresAt: licenseService.licensePayload.expiresAt,
remoteStatus: licenseService.cachedStatus?.status || 'UNKNOWN',
gracePeriodEnds: licenseService.cachedStatus?.gracePeriodEnds,
activeSeats: activeSeats,
isExpired: new Date(licenseService.licensePayload.expiresAt) < new Date(),
features: features,
};
const response: ConsolidatedLicenseStatus = {
customerName: licenseService.licensePayload.customerName,
planSeats: licenseService.licensePayload.planSeats,
expiresAt: licenseService.licensePayload.expiresAt,
remoteStatus: licenseService.cachedStatus?.status || 'UNKNOWN',
gracePeriodEnds: licenseService.cachedStatus?.gracePeriodEnds,
activeSeats: activeSeats,
isExpired: new Date(licenseService.licensePayload.expiresAt) < new Date(),
features: features,
};
res.status(200).json(response);
};
res.status(200).json(response);
};
}
export const licenseController = new LicenseController();

8
packages/enterprise/src/modules/license/license.module.ts
View File

@@ -4,11 +4,11 @@ import { licenseRoutes } from './license.routes';
import { OpenArchiverFeature } from '@open-archiver/types';
class LicenseModule implements ArchiverModule {
name: OpenArchiverFeature = OpenArchiverFeature.STATUS;
name: OpenArchiverFeature = OpenArchiverFeature.STATUS;
async initialize(app: Express, authService: AuthService): Promise<void> {
app.use(`/${config.api.version}/enterprise/status`, licenseRoutes(authService));
}
async initialize(app: Express, authService: AuthService): Promise<void> {
app.use(`/${config.api.version}/enterprise/status`, licenseRoutes(authService));
}
}
export const licenseModule = new LicenseModule();

10
packages/enterprise/src/modules/license/license.routes.ts
View File

@@ -3,12 +3,10 @@ import { licenseController } from './license.controller';
import { requireAuth, AuthService } from '@open-archiver/backend';
export const licenseRoutes = (authService: AuthService): Router => {
const router = Router();
router.use(requireAuth(authService));
const router = Router();
router.use(requireAuth(authService));
router.get(
'/license-status'
, licenseController.getLicenseStatus);
router.get('/license-status', licenseController.getLicenseStatus);
return router;
return router;
};

11
packages/enterprise/src/modules/retention-policy/retention-policy.module.ts
View File

@@ -4,11 +4,14 @@ import { OpenArchiverFeature } from '@open-archiver/types';
import { retentionPolicyRoutes } from './retention-policy.routes';
class RetentionPolicyModule implements ArchiverModule {
name: OpenArchiverFeature = OpenArchiverFeature.RETENTION_POLICY
name: OpenArchiverFeature = OpenArchiverFeature.RETENTION_POLICY;
async initialize(app: Express, authService: AuthService): Promise<void> {
app.use(`/${config.api.version}/enterprise/retention-policy`, retentionPolicyRoutes(authService));
}
async initialize(app: Express, authService: AuthService): Promise<void> {
app.use(
`/${config.api.version}/enterprise/retention-policy`,
retentionPolicyRoutes(authService)
);
}
}
export const retentionPolicyModule = new RetentionPolicyModule();

16
packages/enterprise/src/modules/retention-policy/retention-policy.routes.ts
View File

@@ -4,15 +4,15 @@ import { featureEnabled } from '../../middleware/featureEnabled';
import { OpenArchiverFeature } from '@open-archiver/types';
export const retentionPolicyRoutes = (authService: AuthService): Router => {
const router = Router();
const router = Router();
// All routes in this module require authentication and the retention-policy feature
router.use(requireAuth(authService), featureEnabled(OpenArchiverFeature.RETENTION_POLICY));
// All routes in this module require authentication and the retention-policy feature
router.use(requireAuth(authService), featureEnabled(OpenArchiverFeature.RETENTION_POLICY));
// demonstrating route
router.get('/', (req, res) => {
res.status(200).json({ message: 'Retention policy feature is enabled.' });
});
// demonstrating route
router.get('/', (req, res) => {
res.status(200).json({ message: 'Retention policy feature is enabled.' });
});
return router;
return router;
};

2
packages/frontend/src/app.d.ts vendored
View File

@@ -16,4 +16,4 @@ declare global {
}
}
export { };
export {};

4
packages/frontend/src/hooks.server.ts
View File

@@ -23,9 +23,9 @@ export const handle: Handle = async ({ event, resolve }) => {
event.locals.accessToken = null;
}
if (import.meta.env.VITE_ENTERPRISE_MODE === true) {
event.locals.enterpriseMode = true
event.locals.enterpriseMode = true;
} else {
event.locals.enterpriseMode = false
event.locals.enterpriseMode = false;
}
return resolve(event);

5
packages/frontend/src/routes/api/[...slug]/+server.ts
View File

@@ -22,7 +22,10 @@ const handleRequest: RequestHandler = async ({ request, params, fetch }) => {
return response;
} catch (error) {
console.error('Proxy request failed:', error);
return json({ message: `Failed to connect to the backend service. ${JSON.stringify(error)}` }, { status: 500 });
return json(
{ message: `Failed to connect to the backend service. ${JSON.stringify(error)}` },
{ status: 500 }
);
}
};

31
packages/frontend/src/routes/dashboard/admin/jobs/+page.server.ts
View File

@@ -4,21 +4,24 @@ import type { PageServerLoad } from './$types';
import type { IGetQueuesResponse } from '@open-archiver/types';
export const load: PageServerLoad = async (event) => {
try {
const response = await api('/jobs/queues', event);
try {
const response = await api('/jobs/queues', event);
if (!response.ok) {
const responseText = await response.json();
throw error(response.status as NumericRange<400, 599>, responseText.message || 'Failed to fetch job queues.');
}
if (!response.ok) {
const responseText = await response.json();
throw error(
response.status as NumericRange<400, 599>,
responseText.message || 'Failed to fetch job queues.'
);
}
const data: IGetQueuesResponse = await response.json();
const data: IGetQueuesResponse = await response.json();
return {
queues: data.queues,
};
} catch (e: any) {
console.error('Failed to load job queues:', e);
throw error(e.status || 500, e.body?.message || 'Failed to load job queues');
}
return {
queues: data.queues,
};
} catch (e: any) {
console.error('Failed to load job queues:', e);
throw error(e.status || 500, e.body?.message || 'Failed to load job queues');
}
};

48
packages/frontend/src/routes/dashboard/admin/jobs/[queueName]/+page.server.ts
View File

@@ -4,32 +4,32 @@ import type { PageServerLoad } from './$types';
import type { IGetQueueJobsResponse, JobStatus } from '@open-archiver/types';
export const load: PageServerLoad = async (event) => {
const { queueName } = event.params;
const status = (event.url.searchParams.get('status') || 'failed') as JobStatus;
const page = event.url.searchParams.get('page') || '1';
const limit = event.url.searchParams.get('limit') || '10';
const { queueName } = event.params;
const status = (event.url.searchParams.get('status') || 'failed') as JobStatus;
const page = event.url.searchParams.get('page') || '1';
const limit = event.url.searchParams.get('limit') || '10';
try {
const response = await api(
`/jobs/queues/${queueName}?status=${status}&page=${page}&limit=${limit}`,
event
);
try {
const response = await api(
`/jobs/queues/${queueName}?status=${status}&page=${page}&limit=${limit}`,
event
);
if (!response.ok) {
const responseText = await response.json();
throw error(
response.status as NumericRange<400, 599>,
responseText.message || 'Failed to fetch job queue details.'
);
}
if (!response.ok) {
const responseText = await response.json();
throw error(
response.status as NumericRange<400, 599>,
responseText.message || 'Failed to fetch job queue details.'
);
}
const data: IGetQueueJobsResponse = await response.json();
const data: IGetQueueJobsResponse = await response.json();
return {
queue: data,
};
} catch (e: any) {
console.error('Failed to load job queue details:', e);
throw error(e.status || 500, e.body?.message || 'Failed to load job queue details');
}
return {
queue: data,
};
} catch (e: any) {
console.error('Failed to load job queue details:', e);
throw error(e.status || 500, e.body?.message || 'Failed to load job queue details');
}
};

54
packages/frontend/src/routes/dashboard/admin/license/+page.server.ts
View File

@@ -4,30 +4,36 @@ import type { ConsolidatedLicenseStatus } from '@open-archiver/types';
import { error } from '@sveltejs/kit';
export const load: PageServerLoad = async (event) => {
if (!event.locals.enterpriseMode) {
throw error(403, "This feature is only available in the Enterprise Edition. Please contact Open Archiver to upgrade.")
}
try {
const response = await api('/enterprise/status/license-status', event);
const responseText = await response.json()
if (!response.ok) {
if (response.status === 404) {
throw error(404, responseText.error || JSON.stringify(responseText));
}
// Handle other potential server errors
throw error(response.status, 'Failed to fetch license status');
}
if (!event.locals.enterpriseMode) {
throw error(
403,
'This feature is only available in the Enterprise Edition. Please contact Open Archiver to upgrade.'
);
}
try {
const response = await api('/enterprise/status/license-status', event);
const responseText = await response.json();
if (!response.ok) {
if (response.status === 404) {
throw error(404, responseText.error || JSON.stringify(responseText));
}
// Handle other potential server errors
throw error(response.status, 'Failed to fetch license status');
}
const licenseStatus: ConsolidatedLicenseStatus = responseText
const licenseStatus: ConsolidatedLicenseStatus = responseText;
return {
licenseStatus
};
} catch (e) {
// Catch fetch errors or re-throw kit errors
if (e instanceof Error) {
throw error(500, 'An unexpected error occurred while trying to fetch the license status.');
}
throw e;
}
return {
licenseStatus,
};
} catch (e) {
// Catch fetch errors or re-throw kit errors
if (e instanceof Error) {
throw error(
500,
'An unexpected error occurred while trying to fetch the license status.'
);
}
throw e;
}
};

36
packages/frontend/src/routes/dashboard/compliance/audit-log/+page.server.ts
View File

@@ -4,22 +4,26 @@ import type { PageServerLoad } from './$types';
import type { GetAuditLogsResponse } from '@open-archiver/types';
import { error } from '@sveltejs/kit';
export const load: PageServerLoad = async (event) => {
if (!event.locals.enterpriseMode) {
throw error(403, "This feature is only available in the Enterprise Edition. Please contact Open Archiver to upgrade.")
}
// Forward search params from the page URL to the API request
const response = await api(`/enterprise/audit-logs?${event.url.searchParams.toString()}`, event);
const res = await response.json();
if (!response.ok) {
throw error(response.status, res.message || JSON.stringify(res))
}
const result: GetAuditLogsResponse = res;
return {
logs: result.data,
meta: result.meta
};
if (!event.locals.enterpriseMode) {
throw error(
403,
'This feature is only available in the Enterprise Edition. Please contact Open Archiver to upgrade.'
);
}
// Forward search params from the page URL to the API request
const response = await api(
`/enterprise/audit-logs?${event.url.searchParams.toString()}`,
event
);
const res = await response.json();
if (!response.ok) {
throw error(response.status, res.message || JSON.stringify(res));
}
const result: GetAuditLogsResponse = res;
return {
logs: result.data,
meta: result.meta,
};
};

4
packages/frontend/vite.config.ts
View File

@@ -9,7 +9,7 @@ export default defineConfig({
plugins: [tailwindcss(), sveltekit()],
define: {
// This will be 'true' only during the enterprise build process
'import.meta.env.VITE_ENTERPRISE_MODE': process.env.VITE_ENTERPRISE_MODE === 'true'
'import.meta.env.VITE_ENTERPRISE_MODE': process.env.VITE_ENTERPRISE_MODE === 'true',
},
server: {
port: Number(process.env.PORT_FRONTEND) || 3000,
@@ -19,7 +19,7 @@ export default defineConfig({
changeOrigin: true,
rewrite: (path) => path.replace(/^\/api/, ''),
},
}
},
},
ssr: {
noExternal: ['layerchart'],

52
packages/types/src/audit-log.enums.ts
View File

@@ -1,36 +1,36 @@
export const AuditLogActions = [
// General CRUD
'CREATE',
'READ',
'UPDATE',
'DELETE',
// General CRUD
'CREATE',
'READ',
'UPDATE',
'DELETE',
// User & Session Management
'LOGIN',
'LOGOUT',
'SETUP', // Initial user setup
// User & Session Management
'LOGIN',
'LOGOUT',
'SETUP', // Initial user setup
// Ingestion Actions
'IMPORT',
'PAUSE',
'SYNC',
'UPLOAD',
// Ingestion Actions
'IMPORT',
'PAUSE',
'SYNC',
'UPLOAD',
// Other Actions
'SEARCH',
'DOWNLOAD',
'GENERATE' // For API keys
// Other Actions
'SEARCH',
'DOWNLOAD',
'GENERATE', // For API keys
] as const;
export const AuditLogTargetTypes = [
'ApiKey',
'ArchivedEmail',
'Dashboard',
'IngestionSource',
'Role',
'SystemSettings',
'User',
'File' // For uploads and downloads
'ApiKey',
'ArchivedEmail',
'Dashboard',
'IngestionSource',
'Role',
'SystemSettings',
'User',
'File', // For uploads and downloads
] as const;
export type AuditLogAction = (typeof AuditLogActions)[number];

52
packages/types/src/audit-log.types.ts
View File

@@ -1,39 +1,39 @@
import type { AuditLogAction, AuditLogTargetType } from './audit-log.enums';
export interface AuditLogEntry {
id: number;
previousHash: string | null;
timestamp: Date;
actorIdentifier: string;
actorIp: string | null;
actionType: AuditLogAction;
targetType: AuditLogTargetType | null;
targetId: string | null;
details: Record<string, any> | null;
currentHash: string;
id: number;
previousHash: string | null;
timestamp: Date;
actorIdentifier: string;
actorIp: string | null;
actionType: AuditLogAction;
targetType: AuditLogTargetType | null;
targetId: string | null;
details: Record<string, any> | null;
currentHash: string;
}
export type CreateAuditLogEntry = Omit<
AuditLogEntry,
'id' | 'previousHash' | 'timestamp' | 'currentHash'
AuditLogEntry,
'id' | 'previousHash' | 'timestamp' | 'currentHash'
>;
export interface GetAuditLogsOptions {
page?: number;
limit?: number;
startDate?: Date;
endDate?: Date;
actor?: string;
actionType?: AuditLogAction;
targetType?: AuditLogTargetType | null;
sort?: 'asc' | 'desc';
page?: number;
limit?: number;
startDate?: Date;
endDate?: Date;
actor?: string;
actionType?: AuditLogAction;
targetType?: AuditLogTargetType | null;
sort?: 'asc' | 'desc';
}
export interface GetAuditLogsResponse {
data: AuditLogEntry[];
meta: {
total: number;
page: number;
limit: number;
};
data: AuditLogEntry[];
meta: {
total: number;
page: number;
limit: number;
};
}

2
packages/types/src/index.ts
View File

@@ -12,4 +12,4 @@ export * from './audit-log.types';
export * from './audit-log.enums';
export * from './integrity.types';
export * from './jobs.types';
export * from './license.types'
export * from './license.types';

10
packages/types/src/integrity.types.ts
View File

@@ -1,7 +1,7 @@
export interface IntegrityCheckResult {
type: 'email' | 'attachment';
id: string;
filename?: string;
isValid: boolean;
reason?: string;
type: 'email' | 'attachment';
id: string;
filename?: string;
isValid: boolean;
reason?: string;
}

68
packages/types/src/jobs.types.ts
View File

@@ -7,59 +7,59 @@ export type JobStatus = 'active' | 'completed' | 'failed' | 'delayed' | 'waiting
* A detailed representation of a job, providing essential information for monitoring and debugging.
*/
export interface IJob {
id: string | undefined;
name: string;
data: any;
state: string;
failedReason: string | undefined;
timestamp: number;
processedOn: number | undefined;
finishedOn: number | undefined;
attemptsMade: number;
stacktrace: string[];
returnValue: any;
ingestionSourceId?: string;
error?: any;
id: string | undefined;
name: string;
data: any;
state: string;
failedReason: string | undefined;
timestamp: number;
processedOn: number | undefined;
finishedOn: number | undefined;
attemptsMade: number;
stacktrace: string[];
returnValue: any;
ingestionSourceId?: string;
error?: any;
}
/**
* Holds the count of jobs in various states for a single queue.
*/
export interface IQueueCounts {
active: number;
completed: number;
failed: number;
delayed: number;
waiting: number;
paused: number;
active: number;
completed: number;
failed: number;
delayed: number;
waiting: number;
paused: number;
}
/**
* Provides a high-level overview of a queue, including its name and job counts.
*/
export interface IQueueOverview {
name: string;
counts: IQueueCounts;
name: string;
counts: IQueueCounts;
}
/**
* Represents the pagination details for a list of jobs.
*/
export interface IPagination {
currentPage: number;
totalPages: number;
totalJobs: number;
limit: number;
currentPage: number;
totalPages: number;
totalJobs: number;
limit: number;
}
/**
* Provides a detailed view of a specific queue, including a paginated list of its jobs.
*/
export interface IQueueDetails {
name: string;
counts: IQueueCounts;
jobs: IJob[];
pagination: IPagination;
name: string;
counts: IQueueCounts;
jobs: IJob[];
pagination: IPagination;
}
// --- API Request & Response Types ---
@@ -68,23 +68,23 @@ export interface IQueueDetails {
* Response body for the endpoint that lists all queues.
*/
export interface IGetQueuesResponse {
queues: IQueueOverview[];
queues: IQueueOverview[];
}
/**
* URL parameters for the endpoint that retrieves jobs from a specific queue.
*/
export interface IGetQueueJobsRequestParams {
queueName: string;
queueName: string;
}
/**
* Query parameters for filtering and paginating jobs within a queue.
*/
export interface IGetQueueJobsRequestQuery {
status: JobStatus;
page: string; // Received as a string from query params
limit: string; // Received as a string from query params
status: JobStatus;
page: string; // Received as a string from query params
limit: string; // Received as a string from query params
}
/**

52
packages/types/src/license.types.ts
View File

@@ -2,48 +2,48 @@
* Features of Open Archiver Enterprise
*/
export enum OpenArchiverFeature {
AUDIT_LOG = 'audit-log',
RETENTION_POLICY = 'retention-policy',
SSO = 'sso',
STATUS = 'status',
ALL = 'all',
AUDIT_LOG = 'audit-log',
RETENTION_POLICY = 'retention-policy',
SSO = 'sso',
STATUS = 'status',
ALL = 'all',
}
/**
* The payload of the offline license.jwt file.
*/
export interface LicenseFilePayload {
licenseId: string; // UUID linking to the License Server
customerName: string;
planSeats: number;
features: OpenArchiverFeature[];
expiresAt: string; // ISO 8601
issuedAt: string; // ISO 8601
licenseId: string; // UUID linking to the License Server
customerName: string;
planSeats: number;
features: OpenArchiverFeature[];
expiresAt: string; // ISO 8601
issuedAt: string; // ISO 8601
}
/**
* The structure of the cached response from the License Server.
*/
export interface LicenseStatusPayload {
status: 'VALID' | 'REVOKED';
gracePeriodEnds?: string; // ISO 8601, only present if REVOKED
status: 'VALID' | 'REVOKED';
gracePeriodEnds?: string; // ISO 8601, only present if REVOKED
}
/**
* The consolidated license status object returned by the API.
*/
export interface ConsolidatedLicenseStatus {
// From the license.jwt file
customerName: string;
planSeats: number;
expiresAt: string;
// From the cached license-status.json
remoteStatus: 'VALID' | 'REVOKED' | 'UNKNOWN';
gracePeriodEnds?: string;
// Calculated values
activeSeats: number;
isExpired: boolean;
features: {
[key in OpenArchiverFeature]?: boolean;
};
// From the license.jwt file
customerName: string;
planSeats: number;
expiresAt: string;
// From the cached license-status.json
remoteStatus: 'VALID' | 'REVOKED' | 'UNKNOWN';
gracePeriodEnds?: string;
// Calculated values
activeSeats: number;
isExpired: boolean;
features: {
[key in OpenArchiverFeature]?: boolean;
};
}