Open Age Service Framework

This repository contains a collection of essential services and components for building robust Node.js applications following a modular architecture pattern. It provides standardized boilerplate for REST APIs, offline event processing, database connectivity, and telemetry.

Installation

npm install @open-age/service --save

Quick Start

The framework provides three main entry points for different service processes:

1. API Server (bin/api.js)

Starts the REST API server with middleware, routing, and database initialization.

2. Cron Jobs (bin/cron.js)

Executes scheduled tasks and background jobs.

3. Event Listener (bin/listener.js)

Processes queued events from the offline queue system.

Core Components

1. API Framework (@open-age/service).api

Implements complete REST API boilerplate with built-in middleware for authentication, authorization, caching, and response handling.

Key Features

• REST API routing and request/response handling
• JWT-based authentication
• Request validation and error handling
• Response caching with Redis
• Response standardization
• Bulk request processing
• Swagger/OpenAPI documentation generation

Initialization Example

const oa = require('@open-age/service');
const logger = oa.logger('bin/api');

// Set up folder structure
oa.constants.folders.set([
    { name: 'api', folder: path.join(appRoot.path, 'api') },
    { name: 'subscribers', folder: path.join(appRoot.path, 'processors') },
    { name: 'middlewares', folder: path.join(appRoot.path, 'middlewares') },
    { name: 'services', folder: path.join(appRoot.path, 'services') },
    { name: 'mappers', folder: path.join(appRoot.path, 'mappers') },
    { name: 'models', folder: path.join(appRoot.path, 'models') },
    { name: 'specs.paths', folder: path.join(appRoot.path, 'specs', 'paths') },
    { name: 'specs.definitions', folder: path.join(appRoot.path, 'specs', 'definitions') },
    { name: 'public', folder: path.join(appRoot.path, 'public') }
]);

// Initialize database
global.db = await oa.db.init({
    database: dbConfig.database,
    provider: dbConfig.provider,
    models: dbConfig.models
}, logger);

// Initialize API
await oa.api.init({
    web: {
        port: 3000
    },
    api: {
        info: {
            title: 'System Service API',
            version: 'v1'
        },
        host: 'https://api.example.com/system/v1',
        prefix: 'api'
    },
    middlewares: {
        auth: authConfig,
        cache: cacheConfig,
        context: contextConfig
    }
}, logger);

Configuration

Web Server

• port: The port on which the web server listens
• limit: Maximum request body size (e.g., '200mb')

API Info

• title: API name/title
• version: API version (e.g., 'v1', 'beta')
• host: Base URL of the API
• prefix: Route prefix for all endpoints

Authentication (Auth Middleware)

Configure JWT-based authentication:

{
    "auth": {
        "provider": {
            "type": "jwt",
            "config": {
                "secret": "your-secret-key",
                "expiresIn": 1440
            }
        },
        "validate": ["ip"]
    }
}

Response Caching (Cache Middleware)

Configure Redis-based response caching:

{
    "cache": {
        "disabled": false,
        "root": "my-service",
        "provider": {
            "type": "redis/cache",
            "config": {
                "host": "127.0.0.1",
                "port": 6379,
                "options": {
                    "maxmemory-policy": "allkeys-lru",
                    "maxmemory": "1gb"
                }
            }
        }
    }
}

Context Pipeline (Context Middleware)

Enrich requests with contextual data:

{
    attributes: {
        organization: organizationService,
        tenant: tenantService,
        session: {
            get: async (claim, context) => {
                return directory.sessions.get(claim.id, context);
            }
        }
    },
    mapper: (context) => ({
        organization: context.organization,
        tenant: context.tenant,
        user: context.user
    })
}

2. Events System (@open-age/service).events

Convention-based pub/sub implementation using Redis queues for asynchronous event processing. Enables decoupled, scalable event handling across the application.

Key Features

• Redis-based distributed event queue
• Convention-based subscriber discovery
• Multiple subscribers per event
• Context preservation across async boundaries
• Runtime subscriber registration
• Configurable concurrency and timeouts

Initialization Example

const oa = require('@open-age/service');
const logger = oa.logger('events');

const queues = {
    default: {
        name: 'my-service-offline',
        options: {
            removeOnComplete: true,
            removeOnFail: false
        }
    }
};

await oa.events.init({
    disabled: false,
    namespace: 'my-service',
    concurrency: 4,
    timeout: 30 * 60 * 1000,
    pauseOnError: false,
    queues: queues,
    provider: {
        type: 'redis/queue',
        config: {
            host: '127.0.0.1',
            port: 6379
        }
    },
    subscribers: [
        {
            code: 'composer',
            subscribe: async (event, context) => {
                // Handle event globally across all entity types
                await oa.client.composer.tasks.create({
                    type: { code: `${event.entityType}-${event.action}` },
                    data: event.data
                }, context);
            }
        }
    ],
    context: {
        attributes: {
            organization: require('./services/organizations'),
            tenant: require('./services/tenants'),
            session: {
                get: async (claim, context) => {
                    return oa.client.directory.sessions.get(claim.id, context);
                }
            }
        },
        mapper: (context) => ({
            organization: context.organization,
            tenant: context.tenant,
            user: context.user
        })
    }
}, logger);

Queuing Events

Queue events to be processed asynchronously:

// In a request handler or service
const event = {
    entityType: 'user',      // The entity type being acted upon
    action: 'created',       // The action performed
    data: newUserObject      // The entity data
};

await oa.events.queue(event, context);

Creating Subscribers (File Convention)

Create subscriber files in the subscribers/ folder following the convention:

Single subscriber (events/subscribers/{entityType}/{action}.js):

// subscribers/user/created.js
exports.subscribe = async (event, context) => {
    const logger = context.logger.start('sending welcome email');
    // Send welcome email for new user
    logger.end();
};

Multiple subscribers (events/subscribers/{entityType}/{action}/{subscriberName}.js):

// subscribers/user/created/sendEmail.js
exports.subscribe = async (event, context) => {
    await emailService.sendWelcomeEmail(event.data, context);
};

// subscribers/user/created/createProfile.js
exports.subscribe = async (event, context) => {
    await profileService.initialize(event.data, context);
};

Processing Events with a Listener

Create a separate process to listen for and process queued events:

# Terminal 1: Start the listener
node bin/listener.js

# Terminal 2: Start the API (queues events)
node bin/api.js

Listener implementation (bin/listener.js):

const oa = require('@open-age/service');
const offline = oa.events;

// Configure and listen for events
await offline.init(eventConfig, logger);
await offline.listen(process.env.QUEUE_NAME, logger);

Configuration Reference

• disabled: Set to true to process events synchronously (in-process)
• namespace: Prefix for Redis keys (usually the service name)
• concurrency: Number of parallel event processors
• timeout: Maximum processing time per event
• pauseOnError: Whether to pause processing on errors

3. Client (@open-age/service).client

HTTP client library for consuming downstream services and APIs. Provides utility methods for common service interactions.

Available Services

• directory: User and organization directory service
• composer: Task composition and workflow service
• gateway: API gateway
• sales: Sales service
• sendIt: Notification/messaging service
• And others as configured

Usage Example

const oa = require('@open-age/service');

// Get user directory role by key
const role = await oa.client.directory.roles.get(roleKey, context);

// Create a task in composer
await oa.client.composer.tasks.create({
    type: { code: 'user-created' },
    data: { entity: { id: userId } }
}, context);

// Get session information
const session = await oa.client.directory.sessions.get(sessionId, context);

Configuration

{
    "providers": {
        "directory": {
            "url": "http://api.example.com/directory/v1/api",
            "role": {
                "key": "api-key-or-token"
            }
        },
        "composer": {
            "url": "http://api.example.com/composer/v1/api"
        }
    }
}

4. Database (@open-age/service).db

MongoDB connection management and model initialization. Handles database connections and provides access to initialized models.

Initialization Example

const oa = require('@open-age/service');

global.db = await oa.db.init({
    database: 'my-service-db',
    provider: {
        type: 'mongodb',
        config: {
            host: '127.0.0.1',
            port: 27017,
            options: {
                useNewUrlParser: true,
                useUnifiedTopology: true,
                user: 'username',
                pass: 'password',
                authSource: 'admin'
            }
        }
    },
    models: {
        nameFormat: 'camelCase',
        options: {
            timestamps: true,
            usePushEach: true
        }
    }
}, logger);

Using Models

Once initialized, models are accessible via the global db object:

// Create a new document
const user = await db.models.user.create({
    name: 'John Doe',
    email: '[email protected]'
}, context);

// Find documents
const users = await db.models.user.find({ status: 'active' });

// Update a document
await db.models.user.updateOne(
    { _id: userId },
    { status: 'inactive' }
);

// Delete a document
await db.models.user.deleteOne({ _id: userId });

Configuration

• database: Database name
• provider.type: Database provider type (e.g., 'mongodb')
• provider.config.host: Database host
• provider.config.port: Database port
• provider.config.options: Provider-specific options
• models.nameFormat: Model naming convention ('camelCase', 'snake_case')
• models.options: Default options for all models

5. Telemetry (@open-age/service).telemetry

Provides logging and application telemetry. Initialize loggers for different modules and levels.

Logger Configuration

{
    "logger": {
        "level": "info",
        "providers": [
            {
                "type": "console",
                "config": {
                    "handleExceptions": true,
                    "format": {
                        "timestamp": "HH:mm:ss",
                        "json": false,
                        "colorize": {
                            "all": true
                        }
                    }
                }
            }
        ]
    }
}

Usage Example

const oa = require('@open-age/service');

// Create a logger for a specific module
const logger = oa.logger('services/users');

// Log with severity levels
logger.info('User created successfully');
logger.warn('User creation took longer than expected');
logger.error('Failed to create user');

// Log with context tracking
const requestLogger = logger.start('processing user request');
try {
    // do work
    requestLogger.end();
} catch (err) {
    requestLogger.error(err);
}

Levels

• debug: Detailed debugging information
• info: General informational messages
• warn: Warning messages for potentially harmful situations
• error: Error messages for error events

6. Base API (@open-age/service).base.api

Helper for creating standard CRUD API routes with automatic mapping and response handling.

const api = require('@open-age/service').base.api;

// Create a CRUD router for users
const userRouter = api('users', 'user');

// Routes automatically created:
// GET /users
// GET /users/:id
// POST /users
// PUT /users/:id
// DELETE /users/:id

7. Constants (@open-age/service).constants

Folders

Configure application folder structure:

const oa = require('@open-age/service');

// Set folders individually
oa.constants.folders.set('api', path.join(appRoot, 'api'));
oa.constants.folders.set('models', path.join(appRoot, 'models'));

// Or set in bulk
oa.constants.folders.set([
    { name: 'api', folder: path.join(appRoot, 'api') },
    { name: 'services', folder: path.join(appRoot, 'services') },
    { name: 'models', folder: path.join(appRoot, 'models') },
    { name: 'mappers', folder: path.join(appRoot, 'mappers') },
    { name: 'subscribers', folder: path.join(appRoot, 'processors') }
]);

// Get folder paths
const apiFolder = oa.constants.folders.get('api');

Standard Folders

• api: API route handlers
• services: Business logic services
• models: Database models
• mappers: Response/request mappers
• middlewares: Express middleware
• subscribers: Event subscribers
• specs.paths: OpenAPI path definitions
• specs.definitions: OpenAPI schema definitions
• public: Static assets
• uploads: User uploads
• temp: Temporary files

Errors

Standard error types and custom error definitions:

const oa = require('@open-age/service');
const errors = oa.constants.errors;

// Built-in errors
errors.UNKNOWN
errors.ACCESS_DENIED
errors.INVALID_TOKEN
errors.SESSION_EXPIRED
errors.METHOD_NOT_SUPPORTED
errors.INVALID_STRING

// Throw an error
throw errors.ACCESS_DENIED;

// Custom errors
const customError = {
    code: 'CUSTOM_ERROR',
    status: 403,
    message: 'Custom error message'
};

Built-in Error Types

• UNKNOWN: Unknown error
• ACCESS_DENIED: User lacks permission
• INVALID_TOKEN: Authentication token is invalid
• CLAIMS_EXPIRED: Authentication claims have expired
• SESSION_EXPIRED: User session has expired
• INVALID_IP: Request from invalid IP address
• INVALID_DEVICE: Request from unregistered device
• METHOD_NOT_SUPPORTED: HTTP method not allowed
• INVALID_STRING: String validation failed

Application Architecture

The framework supports a multi-process architecture for scalable service applications:

Process Types

1. API Server (bin/api.js)

Main HTTP server handling REST API requests.

Responsibilities:

• Accept HTTP requests
• Validate and authenticate requests
• Execute business logic via services and models
• Queue events for asynchronous processing
• Return responses to clients

Initialization Steps:

// 1. Configure folder structure
oa.constants.folders.set([...]);

// 2. Initialize database
global.db = await oa.db.init({...}, logger);

// 3. Initialize events system
await oa.events.init({...}, logger);

// 4. Initialize HTTP API
await oa.api.init({...}, logger);

2. Event Listener (bin/listener.js)

Dedicated process for consuming and processing queued events.

Responsibilities:

• Listen for events from Redis queue
• Execute subscriber handlers
• Maintain separate event context
• Log and handle processing errors

Implementation:

const offline = require('@open-age/service').events;
await offline.init(config, logger);
await offline.listen(process.env.QUEUE_NAME, logger);

Running:

node bin/listener.js
# or with queue name filter
QUEUE_NAME=default node bin/listener.js

3. Cron Scheduler (bin/cron.js)

Process for executing scheduled tasks and background jobs.

Responsibilities:

• Load job definitions from /jobs folder
• Schedule recurring tasks
• Execute jobs based on schedule
• Support filtering by organization code

Job File Convention:

// jobs/cleanupExpiredTokens.js
exports.schedule = (orgCodes) => {
    // Use node-cron or similar to schedule
    cron.schedule('0 2 * * *', async () => {
        // execute cleanup
    });
};

Running:

# Run all jobs
node bin/cron.js

# Run specific job
CRON_NAME=cleanupExpiredTokens node bin/cron.js

# Run for specific organizations
ORG_CODES=org1,org2 node bin/cron.js

Directory Structure

service/
├── bin/
│   ├── api.js           # API server entry point
│   ├── cron.js          # Cron jobs entry point
│   └── listener.js      # Event listener entry point
├── api/                 # HTTP route handlers
├── models/              # Database models
├── services/            # Business logic
├── mappers/             # Response mappers
├── processors/          # Event subscribers
├── middlewares/         # Express middleware
├── specs/               # OpenAPI specs
│   ├── paths/
│   └── definitions/
├── helpers/             # Utility helpers
├── settings/            # Configuration setup
├── jobs/                # Cron job definitions
└── config/              # Configuration files

Common Patterns

Service Initialization Pattern

Services using this framework typically follow this startup pattern:

// bin/api.js or bin/listener.js
const about = require('../package.json');
process.env.APP = `${about.name}-api`;
const oa = require('@open-age/service');
const logger = oa.logger('bin/api').start(`booting.${process.env.APP}`);

const boot = async () => {
    // 1. Configure constant folders
    oa.constants.folders.set([...]);
    
    // 2. Initialize database
    global.db = await oa.db.init({...}, logger);
    
    // 3. Initialize events
    await oa.events.init({...}, logger);
    
    // 4. Initialize API
    await oa.api.init({...}, logger);
    
    logger.end();
};

boot().catch(err => {
    logger.error(err);
    process.exit(1);
});

Event Processing Pattern

The framework encourages event-driven architecture:

1. API triggers event: When something happens, queue an event

   await oa.events.queue({
       entityType: 'user',
       action: 'created',
       data: user
   }, context);

2. Global subscribers: Process events that apply across the app

   // Global subscriber in api.js init
   {
       code: 'composer',
       subscribe: async (event, context) => {
           await oa.client.composer.tasks.create({...}, context);
       }
   }

3. Type-specific subscribers: Handle entity-specific events

   // processors/user/created.js
   exports.subscribe = async (event, context) => {
       // Send welcome email
   };

Context Enrichment Pattern

Middleware automatically enriches request context with domain data:

context: {
    attributes: {
        // Services that fetch/enrich data
        organization: organizationService,
        tenant: tenantService,
        
        // Custom get/after logic
        session: {
            get: async (claim, context) => {
                return directory.sessions.get(claim.id, context);
            },
            after: async (session, context) => {
                // Additional processing
            }
        }
    },
    
    // What gets serialized across async boundaries
    mapper: (context) => ({
        organization: context.organization,
        tenant: context.tenant,
        user: context.user
    })
}

Best Practices

1. Separate Concerns: Use events to decouple API logic from background processing
2. Context Propagation: Always pass context through async operations for consistent logging and user tracking
3. Error Handling: Use framework error constants for consistent error responses
4. Configuration: Use environment-specific config files (config/default.json, config/beta.json)
5. Logging: Create loggers per module for better debugging and monitoring
6. Database Models: Leverage timestamps and naming conventions for consistency
7. API Routes: Use mappers to transform models into clean API responses
8. Subscribers: Keep subscribers focused and fast; offload heavy processing to jobs

Contributing

Please read our contributing guidelines before submitting pull requests.

License

This project is licensed under the MIT License - see the LICENSE file for details.