Initial commit with Advoware proxy

This commit is contained in:
root
2025-10-19 14:57:07 +00:00
commit 273aa8b549
45771 changed files with 5534555 additions and 0 deletions

View File

@@ -0,0 +1,21 @@
MIT License
Copyright (c) 2025 Motia
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.

View File

@@ -0,0 +1,124 @@
# @motiadev/core
Core functionality for the Motia framework, providing the foundation for building event-driven workflows.
## Installation
```bash
npm install @motiadev/core
# or
yarn add @motiadev/core
# or
pnpm add @motiadev/core
```
## Overview
`@motiadev/core` is the foundation of the Motia framework, providing:
- Event-driven architecture with pub/sub capabilities
- Multi-language support (TypeScript, Python, Ruby)
- State management
- Cron job scheduling
- API route handling
- Logging infrastructure
## Key Components
### Server
Create and manage an HTTP server for handling API requests:
```typescript
import { createServer } from '@motiadev/core'
const server = createServer(lockedData, eventManager, stateAdapter, config)
```
### Event Management
Publish and subscribe to events across your application:
```typescript
import { createEventManager } from '@motiadev/core'
const eventManager = createEventManager()
// Subscribe to events
eventManager.subscribe({
event: 'user.created',
handlerName: 'sendWelcomeEmail',
filePath: '/path/to/handler.ts',
handler: (event) => {
// Handle the event
},
})
// Emit events
eventManager.emit({
topic: 'user.created',
data: { userId: '123' },
traceId: 'trace-123',
logger: logger,
})
```
### Step Handlers
Create handlers for different types of steps (API, Event, Cron):
```typescript
import { createStepHandlers } from '@motiadev/core'
const stepHandlers = createStepHandlers(lockedData, eventManager, state, config)
```
### State Management
Manage application state with different adapters:
```typescript
import { createStateAdapter } from '@motiadev/core'
const stateAdapter = createStateAdapter({
adapter: 'redis',
host: 'localhost',
port: 6379,
})
// Use state in your handlers
await state.set(traceId, 'key', value)
const value = await state.get(traceId, 'key')
```
### Cron Jobs
Schedule and manage cron jobs:
```typescript
import { setupCronHandlers } from '@motiadev/core'
const cronManager = setupCronHandlers(lockedData, eventManager, state, loggerFactory)
```
## Multi-language Support
Motia supports writing step handlers in multiple languages:
- TypeScript/JavaScript
- Python
- Ruby
Each language has its own runner that communicates with the core framework.
## Types
The package exports TypeScript types for all components:
```typescript
import { Event, FlowContext, ApiRouteConfig, EventConfig, CronConfig } from '@motiadev/core'
```
## License
This package is part of the Motia framework and is licensed under the same terms.

View File

@@ -0,0 +1,17 @@
export * from './src/types';
export { createServer, MotiaServer } from './src/server';
export { createStepHandlers, MotiaEventManager } from './src/step-handlers';
export { createEventManager } from './src/event-manager';
export { Logger } from './src/logger';
export { createStateAdapter } from './src/state/create-state-adapter';
export { setupCronHandlers, CronManager } from './src/cron-handler';
export { isApiStep, isCronStep, isEventStep, isNoopStep } from './src/guards';
export { LockedData } from './src/locked-data';
export { getStepConfig, getStreamConfig } from './src/get-step-config';
export { StateAdapter } from './src/state/state-adapter';
export { createMermaidGenerator } from './src/mermaid-generator';
export { StreamConfig, MotiaStream } from './src/types-stream';
export { getProjectIdentifier, getUserIdentifier, isAnalyticsEnabled, trackEvent } from './src/analytics/utils';
export { Motia } from './src/motia';
export { NoPrinter, Printer } from './src/printer';
export { NoTracer } from './src/observability/no-tracer';

View File

@@ -0,0 +1,52 @@
"use strict";
var __createBinding = (this && this.__createBinding) || (Object.create ? (function(o, m, k, k2) {
if (k2 === undefined) k2 = k;
var desc = Object.getOwnPropertyDescriptor(m, k);
if (!desc || ("get" in desc ? !m.__esModule : desc.writable || desc.configurable)) {
desc = { enumerable: true, get: function() { return m[k]; } };
}
Object.defineProperty(o, k2, desc);
}) : (function(o, m, k, k2) {
if (k2 === undefined) k2 = k;
o[k2] = m[k];
}));
var __exportStar = (this && this.__exportStar) || function(m, exports) {
for (var p in m) if (p !== "default" && !Object.prototype.hasOwnProperty.call(exports, p)) __createBinding(exports, m, p);
};
Object.defineProperty(exports, "__esModule", { value: true });
exports.NoTracer = exports.Printer = exports.NoPrinter = exports.trackEvent = exports.isAnalyticsEnabled = exports.getUserIdentifier = exports.getProjectIdentifier = exports.createMermaidGenerator = exports.getStreamConfig = exports.getStepConfig = exports.LockedData = exports.isNoopStep = exports.isEventStep = exports.isCronStep = exports.isApiStep = exports.setupCronHandlers = exports.createStateAdapter = exports.Logger = exports.createEventManager = exports.createStepHandlers = exports.createServer = void 0;
__exportStar(require("./src/types"), exports);
var server_1 = require("./src/server");
Object.defineProperty(exports, "createServer", { enumerable: true, get: function () { return server_1.createServer; } });
var step_handlers_1 = require("./src/step-handlers");
Object.defineProperty(exports, "createStepHandlers", { enumerable: true, get: function () { return step_handlers_1.createStepHandlers; } });
var event_manager_1 = require("./src/event-manager");
Object.defineProperty(exports, "createEventManager", { enumerable: true, get: function () { return event_manager_1.createEventManager; } });
var logger_1 = require("./src/logger");
Object.defineProperty(exports, "Logger", { enumerable: true, get: function () { return logger_1.Logger; } });
var create_state_adapter_1 = require("./src/state/create-state-adapter");
Object.defineProperty(exports, "createStateAdapter", { enumerable: true, get: function () { return create_state_adapter_1.createStateAdapter; } });
var cron_handler_1 = require("./src/cron-handler");
Object.defineProperty(exports, "setupCronHandlers", { enumerable: true, get: function () { return cron_handler_1.setupCronHandlers; } });
var guards_1 = require("./src/guards");
Object.defineProperty(exports, "isApiStep", { enumerable: true, get: function () { return guards_1.isApiStep; } });
Object.defineProperty(exports, "isCronStep", { enumerable: true, get: function () { return guards_1.isCronStep; } });
Object.defineProperty(exports, "isEventStep", { enumerable: true, get: function () { return guards_1.isEventStep; } });
Object.defineProperty(exports, "isNoopStep", { enumerable: true, get: function () { return guards_1.isNoopStep; } });
var locked_data_1 = require("./src/locked-data");
Object.defineProperty(exports, "LockedData", { enumerable: true, get: function () { return locked_data_1.LockedData; } });
var get_step_config_1 = require("./src/get-step-config");
Object.defineProperty(exports, "getStepConfig", { enumerable: true, get: function () { return get_step_config_1.getStepConfig; } });
Object.defineProperty(exports, "getStreamConfig", { enumerable: true, get: function () { return get_step_config_1.getStreamConfig; } });
var mermaid_generator_1 = require("./src/mermaid-generator");
Object.defineProperty(exports, "createMermaidGenerator", { enumerable: true, get: function () { return mermaid_generator_1.createMermaidGenerator; } });
var utils_1 = require("./src/analytics/utils");
Object.defineProperty(exports, "getProjectIdentifier", { enumerable: true, get: function () { return utils_1.getProjectIdentifier; } });
Object.defineProperty(exports, "getUserIdentifier", { enumerable: true, get: function () { return utils_1.getUserIdentifier; } });
Object.defineProperty(exports, "isAnalyticsEnabled", { enumerable: true, get: function () { return utils_1.isAnalyticsEnabled; } });
Object.defineProperty(exports, "trackEvent", { enumerable: true, get: function () { return utils_1.trackEvent; } });
var printer_1 = require("./src/printer");
Object.defineProperty(exports, "NoPrinter", { enumerable: true, get: function () { return printer_1.NoPrinter; } });
Object.defineProperty(exports, "Printer", { enumerable: true, get: function () { return printer_1.Printer; } });
var no_tracer_1 = require("./src/observability/no-tracer");
Object.defineProperty(exports, "NoTracer", { enumerable: true, get: function () { return no_tracer_1.NoTracer; } });

View File

@@ -0,0 +1,6 @@
export let roots: string[];
export let transform: {
'^.+\\.ts$': string;
};
export let testRegex: string;
export let moduleFileExtensions: string[];

View File

@@ -0,0 +1,5 @@
export declare const getProjectName: (baseDir: string) => string;
export declare const getUserIdentifier: () => string;
export declare const getProjectIdentifier: (baseDir: string) => string;
export declare const isAnalyticsEnabled: () => boolean;
export declare const trackEvent: (eventName: string, properties?: Record<string, any>) => void;

View File

@@ -0,0 +1,52 @@
"use strict";
var __importDefault = (this && this.__importDefault) || function (mod) {
return (mod && mod.__esModule) ? mod : { "default": mod };
};
Object.defineProperty(exports, "__esModule", { value: true });
exports.trackEvent = exports.isAnalyticsEnabled = exports.getProjectIdentifier = exports.getUserIdentifier = exports.getProjectName = void 0;
const path_1 = __importDefault(require("path"));
const fs_1 = __importDefault(require("fs"));
const os_1 = __importDefault(require("os"));
const crypto_1 = __importDefault(require("crypto"));
const analytics_node_1 = require("@amplitude/analytics-node");
const getProjectName = (baseDir) => {
const packageJsonPath = path_1.default.join(baseDir, 'package.json');
if (fs_1.default.existsSync(packageJsonPath)) {
const packageJson = JSON.parse(fs_1.default.readFileSync(packageJsonPath, 'utf8'));
return packageJson.name || path_1.default.basename(baseDir);
}
return 'unknown';
};
exports.getProjectName = getProjectName;
const getUserIdentifier = () => {
const userInfo = `${os_1.default.userInfo().username}${os_1.default.hostname()}`;
return crypto_1.default.createHash('sha256').update(userInfo).digest('hex').substring(0, 16);
};
exports.getUserIdentifier = getUserIdentifier;
const getProjectIdentifier = (baseDir) => {
try {
return crypto_1.default.createHash('sha256').update((0, exports.getProjectName)(baseDir)).digest('hex').substring(0, 16);
}
catch (error) {
return 'unknown';
}
};
exports.getProjectIdentifier = getProjectIdentifier;
const isAnalyticsEnabled = () => {
return process.env.MOTIA_ANALYTICS_DISABLED !== 'true';
};
exports.isAnalyticsEnabled = isAnalyticsEnabled;
// eslint-disable-next-line @typescript-eslint/no-explicit-any
const trackEvent = (eventName, properties = {}) => {
try {
if ((0, exports.isAnalyticsEnabled)()) {
(0, analytics_node_1.track)(eventName, properties, {
user_id: (0, exports.getUserIdentifier)() || 'unknown',
});
}
}
catch (error) {
// Silently fail to not disrupt dev server
}
};
exports.trackEvent = trackEvent;

View File

@@ -0,0 +1,14 @@
import { Motia } from './motia';
import { Step } from './types';
import { Logger } from './logger';
import { Tracer } from './observability';
type CallStepFileOptions = {
step: Step;
traceId: string;
data?: unknown;
contextInFirstArg?: boolean;
logger: Logger;
tracer: Tracer;
};
export declare const callStepFile: <TData>(options: CallStepFileOptions, motia: Motia) => Promise<TData | undefined>;
export {};

View File

@@ -0,0 +1,194 @@
"use strict";
var __importDefault = (this && this.__importDefault) || function (mod) {
return (mod && mod.__esModule) ? mod : { "default": mod };
};
Object.defineProperty(exports, "__esModule", { value: true });
exports.callStepFile = void 0;
const path_1 = __importDefault(require("path"));
const utils_1 = require("./analytics/utils");
const process_manager_1 = require("./process-communication/process-manager");
const utils_2 = require("./utils");
const getLanguageBasedRunner = (stepFilePath = '') => {
const isPython = stepFilePath.endsWith('.py');
const isRuby = stepFilePath.endsWith('.rb');
const isNode = stepFilePath.endsWith('.js') || stepFilePath.endsWith('.ts');
if (isPython) {
const pythonRunner = path_1.default.join(__dirname, 'python', 'python-runner.py');
return { runner: pythonRunner, command: 'python', args: [] };
}
else if (isRuby) {
const rubyRunner = path_1.default.join(__dirname, 'ruby', 'ruby-runner.rb');
return { runner: rubyRunner, command: 'ruby', args: [] };
}
else if (isNode) {
if (process.env._MOTIA_TEST_MODE === 'true') {
const nodeRunner = path_1.default.join(__dirname, 'node', 'node-runner.ts');
return { runner: nodeRunner, command: 'node', args: ['-r', 'ts-node/register'] };
}
const nodeRunner = path_1.default.join(__dirname, 'node', 'node-runner.js');
return { runner: nodeRunner, command: 'node', args: [] };
}
throw Error(`Unsupported file extension ${stepFilePath}`);
};
const callStepFile = (options, motia) => {
const { step, traceId, data, tracer, logger, contextInFirstArg = false } = options;
const flows = step.config.flows;
return new Promise((resolve, reject) => {
const streamConfig = motia.lockedData.getStreams();
const streams = Object.keys(streamConfig).map((name) => ({ name }));
const jsonData = JSON.stringify({ data, flows, traceId, contextInFirstArg, streams });
const { runner, command, args } = getLanguageBasedRunner(step.filePath);
let result;
const processManager = new process_manager_1.ProcessManager({
command,
args: [...args, runner, step.filePath, jsonData],
logger,
context: 'StepExecution',
projectRoot: motia.lockedData.baseDir,
});
(0, utils_1.trackEvent)('step_execution_started', {
stepName: step.config.name,
language: command,
type: step.config.type,
streams: streams.length,
});
processManager
.spawn()
.then(() => {
processManager.handler('close', async (err) => {
processManager.kill();
if (err) {
(0, utils_1.trackEvent)('step_execution_error', {
stepName: step.config.name,
traceId,
message: err.message,
});
}
if (err) {
tracer.end({
message: err.message,
code: err.code,
stack: err.stack?.replace(new RegExp(`${motia.lockedData.baseDir}/`), ''),
});
}
else {
tracer.end();
}
});
processManager.handler('log', async (input) => logger.log(input));
processManager.handler('state.get', async (input) => {
tracer.stateOperation('get', input);
return motia.state.get(input.traceId, input.key);
});
processManager.handler('state.set', async (input) => {
tracer.stateOperation('set', { traceId: input.traceId, key: input.key, value: true });
return motia.state.set(input.traceId, input.key, input.value);
});
processManager.handler('state.delete', async (input) => {
tracer.stateOperation('delete', input);
return motia.state.delete(input.traceId, input.key);
});
processManager.handler('state.clear', async (input) => {
tracer.stateOperation('clear', input);
return motia.state.clear(input.traceId);
});
processManager.handler(`state.getGroup`, (input) => {
tracer.stateOperation('getGroup', input);
return motia.state.getGroup(input.groupId);
});
processManager.handler('result', async (input) => {
result = input;
});
processManager.handler('emit', async (input) => {
const flows = step.config.flows;
if (!(0, utils_2.isAllowedToEmit)(step, input.topic)) {
tracer.emitOperation(input.topic, input.data, false);
return motia.printer.printInvalidEmit(step, input.topic);
}
tracer.emitOperation(input.topic, input.data, true);
return motia.eventManager.emit({ ...input, traceId, flows, logger, tracer }, step.filePath);
});
Object.entries(streamConfig).forEach(([name, streamFactory]) => {
const stateStream = streamFactory();
processManager.handler(`streams.${name}.get`, async (input) => {
tracer.streamOperation(name, 'get', input);
return stateStream.get(input.groupId, input.id);
});
processManager.handler(`streams.${name}.set`, async (input) => {
tracer.streamOperation(name, 'set', { groupId: input.groupId, id: input.id, data: true });
return stateStream.set(input.groupId, input.id, input.data);
});
processManager.handler(`streams.${name}.delete`, async (input) => {
tracer.streamOperation(name, 'delete', input);
return stateStream.delete(input.groupId, input.id);
});
processManager.handler(`streams.${name}.getGroup`, async (input) => {
tracer.streamOperation(name, 'getGroup', input);
return stateStream.getGroup(input.groupId);
});
processManager.handler(`streams.${name}.send`, async (input) => {
tracer.streamOperation(name, 'send', input);
return stateStream.send(input.channel, input.event);
});
});
processManager.onStdout((data) => {
try {
const message = JSON.parse(data.toString());
logger.log(message);
}
catch {
logger.info(Buffer.from(data).toString());
}
});
processManager.onStderr((data) => logger.error(Buffer.from(data).toString()));
processManager.onProcessClose((code) => {
processManager.close();
if (code !== 0 && code !== null) {
const error = { message: `Process exited with code ${code}`, code };
tracer.end(error);
(0, utils_1.trackEvent)('step_execution_error', { stepName: step.config.name, traceId, code });
reject(`Process exited with code ${code}`);
}
else {
tracer.end();
resolve(result);
}
});
processManager.onProcessError((error) => {
processManager.close();
tracer.end({
message: error.message,
code: error.code,
stack: error.stack,
});
if (error.code === 'ENOENT') {
(0, utils_1.trackEvent)('step_execution_error', {
stepName: step.config.name,
traceId,
code: error.code,
message: error.message,
});
reject(`Executable ${command} not found`);
}
else {
reject(error);
}
});
})
.catch((error) => {
tracer.end({
message: error.message,
code: error.code,
stack: error.stack,
});
(0, utils_1.trackEvent)('step_execution_error', {
stepName: step.config.name,
traceId,
code: error.code,
message: error.message,
});
reject(`Failed to spawn process: ${error}`);
});
});
};
exports.callStepFile = callStepFile;

View File

@@ -0,0 +1,16 @@
import { StepConfig } from './types';
export type StateConfig = {
adapter: string;
host: string;
port: number;
password?: string;
};
export type Config = {
port: number;
state: StateConfig;
};
export type Step<TConfig extends StepConfig = StepConfig> = {
config: TConfig;
file: string;
filePath: string;
};

View File

@@ -0,0 +1,2 @@
"use strict";
Object.defineProperty(exports, "__esModule", { value: true });

View File

@@ -0,0 +1,12 @@
import { Motia } from './motia';
import { CronConfig, Step } from './types';
export type CronManager = {
createCronJob: (step: Step<CronConfig>) => void;
removeCronJob: (step: Step<CronConfig>) => void;
close: () => void;
};
export declare const setupCronHandlers: (motia: Motia) => {
createCronJob: (step: Step<CronConfig>) => void;
removeCronJob: (step: Step<CronConfig>) => void;
close: () => void;
};

View File

@@ -0,0 +1,89 @@
"use strict";
var __createBinding = (this && this.__createBinding) || (Object.create ? (function(o, m, k, k2) {
if (k2 === undefined) k2 = k;
var desc = Object.getOwnPropertyDescriptor(m, k);
if (!desc || ("get" in desc ? !m.__esModule : desc.writable || desc.configurable)) {
desc = { enumerable: true, get: function() { return m[k]; } };
}
Object.defineProperty(o, k2, desc);
}) : (function(o, m, k, k2) {
if (k2 === undefined) k2 = k;
o[k2] = m[k];
}));
var __setModuleDefault = (this && this.__setModuleDefault) || (Object.create ? (function(o, v) {
Object.defineProperty(o, "default", { enumerable: true, value: v });
}) : function(o, v) {
o["default"] = v;
});
var __importStar = (this && this.__importStar) || (function () {
var ownKeys = function(o) {
ownKeys = Object.getOwnPropertyNames || function (o) {
var ar = [];
for (var k in o) if (Object.prototype.hasOwnProperty.call(o, k)) ar[ar.length] = k;
return ar;
};
return ownKeys(o);
};
return function (mod) {
if (mod && mod.__esModule) return mod;
var result = {};
if (mod != null) for (var k = ownKeys(mod), i = 0; i < k.length; i++) if (k[i] !== "default") __createBinding(result, mod, k[i]);
__setModuleDefault(result, mod);
return result;
};
})();
Object.defineProperty(exports, "__esModule", { value: true });
exports.setupCronHandlers = void 0;
const cron = __importStar(require("node-cron"));
const call_step_file_1 = require("./call-step-file");
const generate_trace_id_1 = require("./generate-trace-id");
const logger_1 = require("./logger");
const setupCronHandlers = (motia) => {
const cronJobs = new Map();
const createCronJob = (step) => {
const { config, filePath } = step;
const { cron: cronExpression, name: stepName, flows } = config;
if (!cron.validate(cronExpression)) {
logger_1.globalLogger.error('[cron handler] invalid cron expression', {
expression: cronExpression,
step: stepName,
});
return;
}
logger_1.globalLogger.debug('[cron handler] setting up cron job', {
filePath,
step: stepName,
cron: cronExpression,
});
const task = cron.schedule(cronExpression, async () => {
const traceId = (0, generate_trace_id_1.generateTraceId)();
const logger = motia.loggerFactory.create({ traceId, flows, stepName });
const tracer = await motia.tracerFactory.createTracer(traceId, step, logger);
try {
await (0, call_step_file_1.callStepFile)({ contextInFirstArg: true, step, traceId, tracer, logger }, motia);
// eslint-disable-next-line @typescript-eslint/no-explicit-any
}
catch (error) {
logger.error('[cron handler] error executing cron job', {
error: error.message,
step: step.config.name,
});
}
});
cronJobs.set(step.filePath, task);
};
const removeCronJob = (step) => {
const task = cronJobs.get(step.filePath);
if (task) {
task.stop();
cronJobs.delete(step.filePath);
}
};
const close = () => {
cronJobs.forEach((task) => task.stop());
cronJobs.clear();
};
motia.lockedData.cronSteps().forEach(createCronJob);
return { createCronJob, removeCronJob, close };
};
exports.setupCronHandlers = setupCronHandlers;

View File

@@ -0,0 +1,2 @@
import { Express } from 'express';
export declare const analyticsEndpoint: (app: Express, baseDir: string) => void;

View File

@@ -0,0 +1,28 @@
"use strict";
Object.defineProperty(exports, "__esModule", { value: true });
exports.analyticsEndpoint = void 0;
const utils_1 = require("../analytics/utils");
const analyticsEndpoint = (app, baseDir) => {
app.get('/motia/analytics/user', (_req, res) => {
const analyticsEnabled = (0, utils_1.isAnalyticsEnabled)();
if (!analyticsEnabled) {
res.json({
userId: null,
projectId: null,
motiaVersion: null,
analyticsEnabled: false,
});
return;
}
res.json({
userId: (0, utils_1.getUserIdentifier)(),
projectId: (0, utils_1.getProjectIdentifier)(baseDir),
motiaVersion: process.env.npm_package_dependencies_motia || 'unknown',
analyticsEnabled: true,
});
});
app.get('/motia/analytics/status', (_req, res) => {
res.json({ analyticsEnabled: (0, utils_1.isAnalyticsEnabled)() });
});
};
exports.analyticsEndpoint = analyticsEndpoint;

View File

@@ -0,0 +1,2 @@
import { LockedData } from '../locked-data';
export declare const apiEndpoints: (lockedData: LockedData) => void;

View File

@@ -0,0 +1,72 @@
"use strict";
Object.defineProperty(exports, "__esModule", { value: true });
exports.apiEndpoints = void 0;
const guards_1 = require("../guards");
const stream_adapter_1 = require("../streams/adapters/stream-adapter");
const mapEndpoint = (step) => {
return {
id: step.filePath,
method: step.config.method,
path: step.config.path,
description: step.config.description,
queryParams: step.config.queryParams,
responseSchema: step.config.responseSchema,
bodySchema: step.config.bodySchema,
flows: step.config.flows,
};
};
class ApiEndpointsStream extends stream_adapter_1.StreamAdapter {
constructor(lockedData) {
super();
this.lockedData = lockedData;
}
async get(id) {
const endpoint = this.lockedData.apiSteps().find((step) => step.config.path === id);
return endpoint ? mapEndpoint(endpoint) : null;
}
async delete(id) {
return { id };
}
async set(_, __, data) {
return data;
}
async getGroup() {
return this.lockedData.apiSteps().map(mapEndpoint);
}
}
const apiEndpoints = (lockedData) => {
const stream = lockedData.createStream({
filePath: '__motia.api-endpoints.ts',
hidden: true,
config: {
name: '__motia.api-endpoints',
baseConfig: { storageType: 'custom', factory: () => new ApiEndpointsStream(lockedData) },
schema: null,
},
})();
const apiStepCreated = (step) => {
if ((0, guards_1.isApiStep)(step)) {
stream.set('default', step.filePath, {
id: step.filePath,
method: step.config.method,
path: step.config.path,
description: step.config.description,
queryParams: step.config.queryParams,
});
}
};
const apiStepUpdated = (step) => {
if ((0, guards_1.isApiStep)(step)) {
stream.set('default', step.filePath, mapEndpoint(step));
}
};
const apiStepRemoved = (step) => {
if ((0, guards_1.isApiStep)(step)) {
stream.delete('default', step.filePath);
}
};
lockedData.onStep('step-created', apiStepCreated);
lockedData.onStep('step-updated', apiStepUpdated);
lockedData.onStep('step-removed', apiStepRemoved);
};
exports.apiEndpoints = apiEndpoints;

View File

@@ -0,0 +1,3 @@
import { Express } from 'express';
import { LockedData } from '../locked-data';
export declare const flowsConfigEndpoint: (app: Express, baseDir: string, lockedData: LockedData) => void;

View File

@@ -0,0 +1,34 @@
"use strict";
var __importDefault = (this && this.__importDefault) || function (mod) {
return (mod && mod.__esModule) ? mod : { "default": mod };
};
Object.defineProperty(exports, "__esModule", { value: true });
exports.flowsConfigEndpoint = void 0;
const path_1 = __importDefault(require("path"));
const zod_1 = require("zod");
const flows_config_stream_1 = require("../streams/flows-config-stream");
const flowsConfigEndpoint = (app, baseDir, lockedData) => {
const configPath = path_1.default.join(baseDir, 'motia-workbench.json');
const stream = new flows_config_stream_1.FlowsConfigStream(configPath);
lockedData.createStream({
filePath: '__motia.flowsConfig',
hidden: true,
config: {
name: '__motia.flowsConfig',
schema: zod_1.z.object({ name: zod_1.z.string(), steps: zod_1.z.any(), edges: zod_1.z.any() }),
baseConfig: { storageType: 'custom', factory: () => stream },
},
}, { disableTypeCreation: true })();
app.post('/__motia/flows/:id/config', async (req, res) => {
const newFlowConfig = req.body;
try {
await stream.set('default', newFlowConfig.id, newFlowConfig);
res.status(200).send({ message: 'Flow config saved successfully' });
}
catch (error) {
console.error('Error saving flow config:', error.message);
res.status(500).json({ error: 'Failed to save flow config' });
}
});
};
exports.flowsConfigEndpoint = flowsConfigEndpoint;

View File

@@ -0,0 +1,2 @@
import { LockedData } from '../locked-data';
export declare const flowsEndpoint: (lockedData: LockedData) => void;

View File

@@ -0,0 +1,29 @@
"use strict";
Object.defineProperty(exports, "__esModule", { value: true });
exports.flowsEndpoint = void 0;
const zod_1 = require("zod");
const flows_stream_1 = require("../streams/flows-stream");
const flows_helper_1 = require("../helper/flows-helper");
const flowsEndpoint = (lockedData) => {
const flowsStream = lockedData.createStream({
filePath: '__motia.flows',
hidden: true,
config: {
name: '__motia.flows',
schema: zod_1.z.object({ id: zod_1.z.string(), name: zod_1.z.string(), steps: zod_1.z.any(), edges: zod_1.z.any() }),
baseConfig: { storageType: 'custom', factory: () => new flows_stream_1.FlowsStream(lockedData) },
},
}, { disableTypeCreation: true })();
lockedData.on('flow-created', (flowId) => {
const flow = lockedData.flows[flowId];
const response = (0, flows_helper_1.generateFlow)(flowId, flow.steps);
flowsStream.set('default', flowId, response);
});
lockedData.on('flow-updated', (flowId) => {
const flow = lockedData.flows[flowId];
const response = (0, flows_helper_1.generateFlow)(flowId, flow.steps);
flowsStream.set('default', flowId, response);
});
lockedData.on('flow-removed', (flowId) => flowsStream.delete('default', flowId));
};
exports.flowsEndpoint = flowsEndpoint;

View File

@@ -0,0 +1,3 @@
import { Express } from 'express';
import { StateAdapter } from '../state/state-adapter';
export declare const stateEndpoints: (app: Express, stateAdapter: StateAdapter) => void;

View File

@@ -0,0 +1,42 @@
"use strict";
Object.defineProperty(exports, "__esModule", { value: true });
exports.stateEndpoints = void 0;
const stateEndpoints = (app, stateAdapter) => {
app.get('/__motia/state', async (req, res) => {
try {
const groupId = req.query.groupId;
const filter = req.query.filter ? JSON.parse(req.query.filter) : undefined;
const items = await stateAdapter.items({ groupId, filter });
res.json(items);
// eslint-disable-next-line @typescript-eslint/no-explicit-any
}
catch (error) {
res.status(500).json({ error: error.message });
}
});
app.post('/__motia/state', async (req, res) => {
try {
const { key, groupId, value } = req.body;
await stateAdapter.set(groupId, key, value);
res.json({ key, groupId, value });
// eslint-disable-next-line @typescript-eslint/no-explicit-any
}
catch (error) {
res.status(500).json({ error: error.message });
}
});
app.post('/__motia/state/delete', async (req, res) => {
try {
for (const id of req.body.ids) {
const [groupId, key] = id.split(':');
await stateAdapter.delete(groupId, key);
}
res.status(204).send();
// eslint-disable-next-line @typescript-eslint/no-explicit-any
}
catch (error) {
res.status(500).json({ error: error.message });
}
});
};
exports.stateEndpoints = stateEndpoints;

View File

@@ -0,0 +1,3 @@
import { Express } from 'express';
import { LockedData } from '../locked-data';
export declare const stepEndpoint: (app: Express, lockedData: LockedData) => void;

View File

@@ -0,0 +1,44 @@
"use strict";
var __importDefault = (this && this.__importDefault) || function (mod) {
return (mod && mod.__esModule) ? mod : { "default": mod };
};
Object.defineProperty(exports, "__esModule", { value: true });
exports.stepEndpoint = void 0;
const promises_1 = __importDefault(require("fs/promises"));
const flows_helper_1 = require("../helper/flows-helper");
const getFeatures = async (filePath) => {
const stat = await promises_1.default.stat(filePath + '-features.json').catch(() => null);
if (!stat || stat.isDirectory()) {
return [];
}
try {
const content = await promises_1.default.readFile(filePath + '-features.json', 'utf8');
return JSON.parse(content);
}
catch (error) {
return [];
}
};
const stepEndpoint = (app, lockedData) => {
app.get('/__motia/step/:stepId', async (req, res) => {
const id = req.params.stepId;
const allSteps = [...lockedData.activeSteps, ...lockedData.devSteps];
const step = allSteps.find((step) => (0, flows_helper_1.generateStepId)(step.filePath) === id);
if (!step) {
res.status(404).send({ error: 'Step not found' });
return;
}
try {
const content = await promises_1.default.readFile(step.filePath, 'utf8');
const features = await getFeatures(step.filePath);
res.status(200).send({ id, content, features });
}
catch (error) {
console.error('Error reading step file:', error);
res.status(500).send({
error: 'Failed to read step file',
});
}
});
};
exports.stepEndpoint = stepEndpoint;

View File

@@ -0,0 +1,3 @@
import { Express } from 'express';
import { TracerFactory } from '../observability';
export declare const traceEndpoint: (app: Express, tracerFactory: TracerFactory) => void;

View File

@@ -0,0 +1,10 @@
"use strict";
Object.defineProperty(exports, "__esModule", { value: true });
exports.traceEndpoint = void 0;
const traceEndpoint = (app, tracerFactory) => {
app.post('/__motia/trace/clear', async (req, res) => {
await tracerFactory.clear();
res.json({ message: 'Traces cleared' });
});
};
exports.traceEndpoint = traceEndpoint;

View File

@@ -0,0 +1,2 @@
import { EventManager } from './types';
export declare const createEventManager: () => EventManager;

View File

@@ -0,0 +1,27 @@
"use strict";
Object.defineProperty(exports, "__esModule", { value: true });
exports.createEventManager = void 0;
const logger_1 = require("./logger");
const createEventManager = () => {
const handlers = {};
const emit = async (event, file) => {
const eventHandlers = handlers[event.topic] ?? [];
const { logger, ...rest } = event;
logger.debug('[Flow Emit] Event emitted', { handlers: eventHandlers.length, data: rest, file });
eventHandlers.map((eventHandler) => eventHandler.handler(event));
};
const subscribe = (config) => {
const { event, handlerName, handler, filePath } = config;
if (!handlers[event]) {
handlers[event] = [];
}
logger_1.globalLogger.debug('[Flow Sub] Subscribing to event', { event, handlerName });
handlers[event].push({ filePath, handler: handler });
};
const unsubscribe = (config) => {
const { filePath, event } = config;
handlers[event] = handlers[event]?.filter((handler) => handler.filePath !== filePath);
};
return { emit, subscribe, unsubscribe };
};
exports.createEventManager = createEventManager;

View File

@@ -0,0 +1,2 @@
export declare const generateCode: (length?: number) => string;
export declare const generateTraceId: () => string;

View File

@@ -0,0 +1,14 @@
"use strict";
Object.defineProperty(exports, "__esModule", { value: true });
exports.generateTraceId = exports.generateCode = void 0;
const generateCode = (length = 6) => {
const chars = 'ABCDEFGHJKMNPQRSTUVWXYZ123456789';
return Array.from({ length }, () => chars[Math.floor(Math.random() * chars.length)]).join('');
};
exports.generateCode = generateCode;
const generateTraceId = () => {
const datePart = String(Date.now()).slice(6);
const randomPart = (0, exports.generateCode)(5);
return `${randomPart}-${datePart}`;
};
exports.generateTraceId = generateTraceId;

View File

@@ -0,0 +1,4 @@
import { StepConfig } from './types';
import { StreamConfig } from './types-stream';
export declare const getStepConfig: (file: string, projectRoot?: string) => Promise<StepConfig | null>;
export declare const getStreamConfig: (file: string, projectRoot?: string) => Promise<StreamConfig | null>;

View File

@@ -0,0 +1,89 @@
"use strict";
var __importDefault = (this && this.__importDefault) || function (mod) {
return (mod && mod.__esModule) ? mod : { "default": mod };
};
Object.defineProperty(exports, "__esModule", { value: true });
exports.getStreamConfig = exports.getStepConfig = void 0;
const path_1 = __importDefault(require("path"));
const logger_1 = require("./logger");
const process_manager_1 = require("./process-communication/process-manager");
const getLanguageBasedRunner = (stepFilePath = '') => {
const isPython = stepFilePath.endsWith('.py');
const isRuby = stepFilePath.endsWith('.rb');
const isNode = stepFilePath.endsWith('.js') || stepFilePath.endsWith('.ts');
if (isPython) {
const pythonRunner = path_1.default.join(__dirname, 'python', 'get-config.py');
return { runner: pythonRunner, command: 'python', args: [] };
}
else if (isRuby) {
const rubyRunner = path_1.default.join(__dirname, 'ruby', 'get-config.rb');
return { runner: rubyRunner, command: 'ruby', args: [] };
}
else if (isNode) {
if (process.env._MOTIA_TEST_MODE === 'true') {
const nodeRunner = path_1.default.join(__dirname, 'node', 'get-config.ts');
return { runner: nodeRunner, command: 'node', args: ['-r', 'ts-node/register'] };
}
const nodeRunner = path_1.default.join(__dirname, 'node', 'get-config.js');
return { runner: nodeRunner, command: 'node', args: [] };
}
throw Error(`Unsupported file extension ${stepFilePath}`);
};
const getConfig = (file, projectRoot) => {
const { runner, command, args } = getLanguageBasedRunner(file);
return new Promise((resolve, reject) => {
let config = null;
const processManager = new process_manager_1.ProcessManager({
command,
args: [...args, runner, file],
logger: logger_1.globalLogger,
context: 'Config',
projectRoot,
});
processManager
.spawn()
.then(() => {
processManager.onMessage((message) => {
config = message;
logger_1.globalLogger.debug(`[Config] Read config via ${processManager.commType?.toUpperCase()}`, {
config,
communicationType: processManager.commType,
});
resolve(config);
processManager.kill();
});
processManager.onProcessClose((code) => {
processManager.close();
if (config) {
return;
}
else if (code !== 0) {
reject(`Process exited with code ${code}`);
}
else if (!config) {
reject(`No config found for file ${file}`);
}
});
processManager.onProcessError((error) => {
processManager.close();
if (error.code === 'ENOENT') {
reject(`Executable ${command} not found`);
}
else {
reject(error);
}
});
})
.catch((error) => {
reject(`Failed to spawn process: ${error}`);
});
});
};
const getStepConfig = (file, projectRoot) => {
return getConfig(file, projectRoot);
};
exports.getStepConfig = getStepConfig;
const getStreamConfig = (file, projectRoot) => {
return getConfig(file, projectRoot);
};
exports.getStreamConfig = getStreamConfig;

View File

@@ -0,0 +1 @@
export declare const getStepLanguage: (fileExtension?: string) => string | undefined;

View File

@@ -0,0 +1,27 @@
"use strict";
Object.defineProperty(exports, "__esModule", { value: true });
exports.getStepLanguage = void 0;
const getStepLanguage = (fileExtension) => {
if (!fileExtension)
return;
if (fileExtension.endsWith('.js')) {
return 'javascript';
}
if (fileExtension.endsWith('.ts')) {
return 'typescript';
}
if (fileExtension.endsWith('.py')) {
return 'python';
}
if (fileExtension.endsWith('.go')) {
return 'go';
}
if (fileExtension.endsWith('.rb')) {
return 'ruby';
}
if (fileExtension.endsWith('.php')) {
return 'php';
}
return;
};
exports.getStepLanguage = getStepLanguage;

View File

@@ -0,0 +1,5 @@
import { ApiRouteConfig, EventConfig, NoopConfig, Step, CronConfig } from './types';
export declare const isApiStep: (step: Step) => step is Step<ApiRouteConfig>;
export declare const isEventStep: (step: Step) => step is Step<EventConfig>;
export declare const isNoopStep: (step: Step) => step is Step<NoopConfig>;
export declare const isCronStep: (step: Step) => step is Step<CronConfig>;

View File

@@ -0,0 +1,11 @@
"use strict";
Object.defineProperty(exports, "__esModule", { value: true });
exports.isCronStep = exports.isNoopStep = exports.isEventStep = exports.isApiStep = void 0;
const isApiStep = (step) => step.config.type === 'api';
exports.isApiStep = isApiStep;
const isEventStep = (step) => step.config.type === 'event';
exports.isEventStep = isEventStep;
const isNoopStep = (step) => step.config.type === 'noop';
exports.isNoopStep = isNoopStep;
const isCronStep = (step) => step.config.type === 'cron';
exports.isCronStep = isCronStep;

View File

@@ -0,0 +1,5 @@
import { Step } from 'src/types';
import { FlowResponse } from '../types/flows-types';
export declare const STEP_NAMESPACE = "7f1c3ff2-9b00-4d0a-bdd7-efb8bca49d4f";
export declare const generateStepId: (filePath: string) => string;
export declare const generateFlow: (flowId: string, flowSteps: Step[]) => FlowResponse;

View File

@@ -0,0 +1,142 @@
"use strict";
var __importDefault = (this && this.__importDefault) || function (mod) {
return (mod && mod.__esModule) ? mod : { "default": mod };
};
Object.defineProperty(exports, "__esModule", { value: true });
exports.generateFlow = exports.generateStepId = exports.STEP_NAMESPACE = void 0;
const guards_1 = require("../guards");
const get_step_language_1 = require("../get-step-language");
const path_1 = __importDefault(require("path"));
const fs_1 = __importDefault(require("fs"));
const uuid_1 = require("uuid");
const getNodeComponentPath = (filePath) => {
const filePathWithoutExtension = filePath.replace(/\.[^/.]+$/, '');
const tsxPath = filePathWithoutExtension + '.tsx';
const jsxPath = filePathWithoutExtension + '.jsx';
if (fs_1.default.existsSync(tsxPath))
return tsxPath;
if (fs_1.default.existsSync(jsxPath))
return jsxPath;
};
const getRelativePath = (filePath) => {
const baseDir = process.cwd();
return path_1.default.relative(baseDir, filePath);
};
const createEdge = (sourceId, targetId, topic, label, variant, conditional) => ({
id: `${sourceId}-${targetId}`,
source: sourceId,
target: targetId,
data: {
variant,
label,
topic,
labelVariant: conditional ? 'conditional' : 'default',
},
});
const processEmit = (emit) => {
const isString = typeof emit === 'string';
return {
topic: isString ? emit : emit.topic,
label: isString ? undefined : emit.label,
conditional: isString ? undefined : emit.conditional,
};
};
const createEdgesForEmits = (sourceStep, targetSteps, emits, variant) => {
const edges = [];
emits.forEach((emit) => {
const { topic, label, conditional } = processEmit(emit);
targetSteps.forEach((targetStep) => {
if (targetStep.subscribes?.includes(topic) || targetStep.virtualSubscribes?.includes(topic)) {
edges.push(createEdge(sourceStep.id, targetStep.id, topic, label, variant, conditional));
}
});
});
return edges;
};
const createBaseStepResponse = (step, id) => ({
id,
name: step.config.name,
description: step.config.description,
nodeComponentPath: getNodeComponentPath(step.filePath),
filePath: getRelativePath(step.filePath),
language: (0, get_step_language_1.getStepLanguage)(step.filePath),
virtualEmits: step.config.virtualEmits ?? undefined,
virtualSubscribes: step.config.virtualSubscribes ?? undefined,
});
const createApiStepResponse = (step, id) => {
if (!(0, guards_1.isApiStep)(step)) {
throw new Error('Attempted to create API step response with non-API step');
}
return {
...createBaseStepResponse(step, id),
type: 'api',
emits: step.config.emits,
subscribes: step.config.virtualSubscribes ?? undefined,
action: 'webhook',
webhookUrl: `${step.config.method} ${step.config.path}`,
bodySchema: step.config.bodySchema ?? undefined,
};
};
const createEventStepResponse = (step, id) => {
if (!(0, guards_1.isEventStep)(step)) {
throw new Error('Attempted to create Event step response with non-Event step');
}
return {
...createBaseStepResponse(step, id),
type: 'event',
emits: step.config.emits,
subscribes: step.config.subscribes,
};
};
const createNoopStepResponse = (step, id) => {
if (!(0, guards_1.isNoopStep)(step)) {
throw new Error('Attempted to create Noop step response with non-Noop step');
}
return {
...createBaseStepResponse(step, id),
type: 'noop',
emits: [],
subscribes: step.config.virtualSubscribes,
};
};
const createCronStepResponse = (step, id) => {
if (!(0, guards_1.isCronStep)(step)) {
throw new Error('Attempted to create Cron step response with non-Cron step');
}
return {
...createBaseStepResponse(step, id),
type: 'cron',
emits: step.config.emits,
cronExpression: step.config.cron,
};
};
exports.STEP_NAMESPACE = '7f1c3ff2-9b00-4d0a-bdd7-efb8bca49d4f';
const generateStepId = (filePath) => {
return (0, uuid_1.v5)(filePath, exports.STEP_NAMESPACE);
};
exports.generateStepId = generateStepId;
const createStepResponse = (step) => {
const id = (0, exports.generateStepId)(step.filePath);
if ((0, guards_1.isApiStep)(step))
return createApiStepResponse(step, id);
if ((0, guards_1.isEventStep)(step))
return createEventStepResponse(step, id);
if ((0, guards_1.isNoopStep)(step))
return createNoopStepResponse(step, id);
if ((0, guards_1.isCronStep)(step))
return createCronStepResponse(step, id);
throw new Error(`Unknown step type for step: ${step.config.name}`);
};
const createEdgesForStep = (sourceStep, allSteps) => {
const regularEdges = createEdgesForEmits(sourceStep, allSteps, sourceStep.emits, 'event');
const virtualEdges = sourceStep.virtualEmits
? createEdgesForEmits(sourceStep, allSteps, sourceStep.virtualEmits, 'virtual')
: [];
return [...regularEdges, ...virtualEdges];
};
const generateFlow = (flowId, flowSteps) => {
const steps = flowSteps.map(createStepResponse);
const edges = steps.flatMap((step) => createEdgesForStep(step, steps));
return { id: flowId, name: flowId, steps, edges };
};
exports.generateFlow = generateFlow;

View File

@@ -0,0 +1,61 @@
import { Printer } from './printer';
import { StreamFactory } from './streams/stream-factory';
import { ApiRouteConfig, CronConfig, EventConfig, Flow, Step } from './types';
import { Stream } from './types-stream';
type FlowEvent = 'flow-created' | 'flow-removed' | 'flow-updated';
type StepEvent = 'step-created' | 'step-removed' | 'step-updated';
type StreamEvent = 'stream-created' | 'stream-removed' | 'stream-updated';
type StreamWrapper<TData> = (streamName: string, factory: StreamFactory<TData>) => StreamFactory<TData>;
export declare class LockedData {
readonly baseDir: string;
readonly streamAdapter: 'file' | 'memory';
private readonly printer;
flows: Record<string, Flow>;
activeSteps: Step[];
devSteps: Step[];
private stepsMap;
private handlers;
private stepHandlers;
private streamHandlers;
private streams;
private streamWrapper?;
constructor(baseDir: string, streamAdapter: "file" | "memory" | undefined, printer: Printer);
applyStreamWrapper<TData>(streamWrapper: StreamWrapper<TData>): void;
saveTypes(): void;
on(event: FlowEvent, handler: (flowName: string) => void): void;
onStep(event: StepEvent, handler: (step: Step) => void): void;
onStream(event: StreamEvent, handler: (stream: Stream) => void): void;
eventSteps(): Step<EventConfig>[];
apiSteps(): Step<ApiRouteConfig>[];
cronSteps(): Step<CronConfig>[];
pythonSteps(): Step[];
tsSteps(): Step[];
getStreams(): Record<string, StreamFactory<any>>;
listStreams(): Stream[];
findStream(path: string): Stream | undefined;
updateStep(oldStep: Step, newStep: Step, options?: {
disableTypeCreation?: boolean;
}): boolean;
createStep(step: Step, options?: {
disableTypeCreation?: boolean;
}): boolean;
deleteStep(step: Step, options?: {
disableTypeCreation?: boolean;
}): void;
private createFactoryWrapper;
createStream<TData>(baseStream: Omit<Stream, 'factory'>, options?: {
disableTypeCreation?: boolean;
}): StreamFactory<TData>;
deleteStream(stream: Stream, options?: {
disableTypeCreation?: boolean;
}): void;
updateStream(oldStream: Stream, stream: Stream, options?: {
disableTypeCreation?: boolean;
}): void;
private createFlow;
private removeFlow;
private onFlowUpdated;
private isValidStep;
private createStreamAdapter;
}
export {};

View File

@@ -0,0 +1,274 @@
"use strict";
var __importDefault = (this && this.__importDefault) || function (mod) {
return (mod && mod.__esModule) ? mod : { "default": mod };
};
Object.defineProperty(exports, "__esModule", { value: true });
exports.LockedData = void 0;
const fs_1 = __importDefault(require("fs"));
const path_1 = __importDefault(require("path"));
const guards_1 = require("./guards");
const step_validator_1 = require("./step-validator");
const file_stream_adapter_1 = require("./streams/adapters/file-stream-adapter");
const memory_stream_adapter_1 = require("./streams/adapters/memory-stream-adapter");
const generate_types_1 = require("./types/generate-types");
class LockedData {
constructor(baseDir, streamAdapter = 'file', printer) {
this.baseDir = baseDir;
this.streamAdapter = streamAdapter;
this.printer = printer;
this.flows = {};
this.activeSteps = [];
this.devSteps = [];
this.stepsMap = {};
this.handlers = {
'flow-created': [],
'flow-removed': [],
'flow-updated': [],
};
this.stepHandlers = {
'step-created': [],
'step-removed': [],
'step-updated': [],
};
this.streamHandlers = {
'stream-created': [],
'stream-removed': [],
'stream-updated': [],
};
this.streams = {};
}
applyStreamWrapper(streamWrapper) {
this.streamWrapper = streamWrapper;
}
saveTypes() {
const types = (0, generate_types_1.generateTypesFromSteps)(this.activeSteps, this.printer);
const streams = (0, generate_types_1.generateTypesFromStreams)(this.streams);
const typesString = (0, generate_types_1.generateTypesString)(types, streams);
fs_1.default.writeFileSync(path_1.default.join(this.baseDir, 'types.d.ts'), typesString);
}
on(event, handler) {
this.handlers[event].push(handler);
}
onStep(event, handler) {
this.stepHandlers[event].push(handler);
}
onStream(event, handler) {
this.streamHandlers[event].push(handler);
}
eventSteps() {
return this.activeSteps.filter(guards_1.isEventStep);
}
apiSteps() {
return this.activeSteps.filter(guards_1.isApiStep);
}
cronSteps() {
return this.activeSteps.filter(guards_1.isCronStep);
}
pythonSteps() {
return this.activeSteps.filter((step) => step.filePath.endsWith('.py'));
}
tsSteps() {
return this.activeSteps.filter((step) => step.filePath.endsWith('.ts'));
}
// eslint-disable-next-line @typescript-eslint/no-explicit-any
getStreams() {
const streams = {};
for (const [key, value] of Object.entries(this.streams)) {
streams[key] = value.factory;
}
return streams;
}
listStreams() {
return Object.values(this.streams);
}
findStream(path) {
return Object.values(this.streams).find((stream) => stream.filePath === path);
}
updateStep(oldStep, newStep, options = {}) {
if (!this.isValidStep(newStep)) {
this.deleteStep(oldStep);
return false;
}
if (oldStep.config.type !== newStep.config.type) {
this.activeSteps = this.activeSteps.filter((s) => s.filePath !== oldStep.filePath);
this.devSteps = this.devSteps.filter((s) => s.filePath !== oldStep.filePath);
if (newStep.config.type === 'noop') {
this.devSteps.push(newStep);
}
else {
this.activeSteps.push(newStep);
}
}
const savedStep = this.stepsMap[newStep.filePath];
const addedFlows = newStep.config.flows?.filter((flowName) => !oldStep.config.flows?.includes(flowName)) ?? [];
const removedFlows = oldStep.config.flows?.filter((flowName) => !newStep.config.flows?.includes(flowName)) ?? [];
const untouchedFlows = oldStep.config.flows?.filter((flowName) => newStep.config.flows?.includes(flowName)) ?? [];
savedStep.config = newStep.config;
untouchedFlows.forEach((flowName) => this.onFlowUpdated(flowName));
for (const flowName of addedFlows) {
if (!this.flows[flowName]) {
const flow = this.createFlow(flowName);
flow.steps.push(savedStep);
}
else {
this.flows[flowName].steps.push(savedStep);
this.onFlowUpdated(flowName);
}
}
for (const flowName of removedFlows) {
const flowSteps = this.flows[flowName].steps;
this.flows[flowName].steps = flowSteps.filter(({ filePath }) => filePath !== newStep.filePath);
if (this.flows[flowName].steps.length === 0) {
this.removeFlow(flowName);
}
else {
this.onFlowUpdated(flowName);
}
}
if (!options.disableTypeCreation) {
this.saveTypes();
}
this.stepHandlers['step-updated'].forEach((handler) => handler(newStep));
this.printer.printStepUpdated(newStep);
return true;
}
createStep(step, options = {}) {
if (!this.isValidStep(step)) {
return false;
}
this.stepsMap[step.filePath] = step;
if (step.config.type === 'noop') {
this.devSteps.push(step);
}
else {
this.activeSteps.push(step);
}
for (const flowName of step.config.flows ?? []) {
if (!this.flows[flowName]) {
const flow = this.createFlow(flowName);
flow.steps.push(step);
}
else {
this.flows[flowName].steps.push(step);
this.onFlowUpdated(flowName);
}
}
if (!options.disableTypeCreation) {
this.saveTypes();
}
this.stepHandlers['step-created'].forEach((handler) => handler(step));
this.printer.printStepCreated(step);
return true;
}
deleteStep(step, options = {}) {
// Remove step from active and dev steps
this.activeSteps = this.activeSteps.filter(({ filePath }) => filePath !== step.filePath);
this.devSteps = this.devSteps.filter(({ filePath }) => filePath !== step.filePath);
delete this.stepsMap[step.filePath];
for (const flowName of step.config.flows ?? []) {
const stepFlows = this.flows[flowName]?.steps;
if (stepFlows) {
this.flows[flowName].steps = stepFlows.filter(({ filePath }) => filePath !== step.filePath);
}
if (this.flows[flowName].steps.length === 0) {
this.removeFlow(flowName);
}
else {
this.onFlowUpdated(flowName);
}
}
if (!options.disableTypeCreation) {
this.saveTypes();
}
this.stepHandlers['step-removed'].forEach((handler) => handler(step));
this.printer.printStepRemoved(step);
}
createFactoryWrapper(stream, factory) {
return () => {
const streamFactory = this.streamWrapper //
? this.streamWrapper(stream.config.name, factory)
: factory;
return streamFactory();
};
}
createStream(baseStream, options = {}) {
const stream = baseStream;
this.streams[stream.config.name] = stream;
this.streamHandlers['stream-created'].forEach((handler) => handler(stream));
if (stream.config.baseConfig.storageType === 'default') {
stream.factory = this.createFactoryWrapper(stream, () => this.createStreamAdapter(stream.config.name));
}
else {
stream.factory = this.createFactoryWrapper(stream, stream.config.baseConfig.factory);
}
if (!stream.hidden) {
this.printer.printStreamCreated(stream);
if (!options.disableTypeCreation) {
this.saveTypes();
}
}
return stream.factory;
}
deleteStream(stream, options = {}) {
Object.entries(this.streams).forEach(([streamName, { filePath }]) => {
if (stream.filePath === filePath) {
delete this.streams[streamName];
}
});
this.streamHandlers['stream-removed'].forEach((handler) => handler(stream));
if (!stream.hidden) {
this.printer.printStreamRemoved(stream);
if (!options.disableTypeCreation) {
this.saveTypes();
}
}
}
updateStream(oldStream, stream, options = {}) {
if (oldStream.config.name !== stream.config.name) {
delete this.streams[oldStream.config.name];
}
if (stream.config.baseConfig.storageType === 'default') {
stream.factory = this.createFactoryWrapper(stream, () => this.createStreamAdapter(stream.config.name));
}
else {
stream.factory = this.createFactoryWrapper(stream, stream.config.baseConfig.factory);
}
this.streams[stream.config.name] = stream;
this.streamHandlers['stream-updated'].forEach((handler) => handler(stream));
if (!stream.hidden) {
this.printer.printStreamUpdated(stream);
if (!options.disableTypeCreation) {
this.saveTypes();
}
}
}
createFlow(flowName) {
const flow = { name: flowName, description: '', steps: [] };
this.flows[flowName] = flow;
this.handlers['flow-created'].forEach((handler) => handler(flowName));
this.printer.printFlowCreated(flowName);
return flow;
}
removeFlow(flowName) {
delete this.flows[flowName];
this.handlers['flow-removed'].forEach((handler) => handler(flowName));
this.printer.printFlowRemoved(flowName);
}
onFlowUpdated(flowName) {
this.handlers['flow-updated'].forEach((handler) => handler(flowName));
}
isValidStep(step) {
const validationResult = (0, step_validator_1.validateStep)(step);
if (!validationResult.success) {
this.printer.printValidationError(step.filePath, validationResult);
}
return validationResult.success;
}
createStreamAdapter(streamName) {
if (this.streamAdapter === 'file') {
return new file_stream_adapter_1.FileStreamAdapter(this.baseDir, streamName);
}
return new memory_stream_adapter_1.MemoryStreamAdapter();
}
}
exports.LockedData = LockedData;

View File

@@ -0,0 +1,18 @@
import { Logger } from './logger';
import { StreamAdapter } from './streams/adapters/stream-adapter';
import { Log } from './streams/logs-stream';
type CreateLogger = {
traceId: string;
flows?: string[];
stepName: string;
};
export interface LoggerFactory {
create: (args: CreateLogger) => Logger;
}
export declare class BaseLoggerFactory implements LoggerFactory {
private readonly isVerbose;
private readonly logStream;
constructor(isVerbose: boolean, logStream: StreamAdapter<Log>);
create({ stepName, traceId, flows }: CreateLogger): Logger;
}
export {};

View File

@@ -0,0 +1,27 @@
"use strict";
Object.defineProperty(exports, "__esModule", { value: true });
exports.BaseLoggerFactory = void 0;
const crypto_1 = require("crypto");
const logger_1 = require("./logger");
class BaseLoggerFactory {
constructor(isVerbose, logStream) {
this.isVerbose = isVerbose;
this.logStream = logStream;
}
create({ stepName, traceId, flows }) {
const streamListener = (level, msg, args) => {
const id = (0, crypto_1.randomUUID)();
this.logStream.set('default', id, {
id,
...(args ?? {}),
level,
time: Date.now(),
msg,
traceId,
flows: flows ?? [],
});
};
return new logger_1.Logger(this.isVerbose, { traceId, flows, step: stepName }, [streamListener]);
}
}
exports.BaseLoggerFactory = BaseLoggerFactory;

View File

@@ -0,0 +1,25 @@
export type LogListener = (level: string, msg: string, args?: unknown) => void;
export declare class Logger {
readonly isVerbose: boolean;
private readonly meta;
private readonly coreListeners;
/**
* Why do we need two level of listeners?
*
* Core listeners pass along to children loggers.
*
* However, base listeners do not pass along to children loggers.
* Those are specific to each logger in the hierarchy.
*/
private readonly listeners;
constructor(isVerbose?: boolean, meta?: Record<string, unknown>, coreListeners?: LogListener[]);
child(meta: Record<string, unknown>): Logger;
private _log;
info(message: string, args?: unknown): void;
error(message: string, args?: unknown): void;
debug(message: string, args?: unknown): void;
warn(message: string, args?: unknown): void;
log(args: any): void;
addListener(listener: LogListener): void;
}
export declare const globalLogger: Logger;

View File

@@ -0,0 +1,62 @@
"use strict";
Object.defineProperty(exports, "__esModule", { value: true });
exports.globalLogger = exports.Logger = void 0;
const pretty_print_1 = require("./pretty-print");
const logLevel = process.env.LOG_LEVEL ?? 'info';
const isDebugEnabled = logLevel === 'debug';
const isInfoEnabled = ['info', 'debug'].includes(logLevel);
const isWarnEnabled = ['warn', 'info', 'debug', 'trace'].includes(logLevel);
class Logger {
constructor(isVerbose = false, meta = {}, coreListeners = []) {
this.isVerbose = isVerbose;
this.meta = meta;
this.coreListeners = coreListeners;
/**
* Why do we need two level of listeners?
*
* Core listeners pass along to children loggers.
*
* However, base listeners do not pass along to children loggers.
* Those are specific to each logger in the hierarchy.
*/
this.listeners = [];
}
child(meta) {
return new Logger(this.isVerbose, { ...this.meta, ...meta }, this.coreListeners);
}
// eslint-disable-next-line @typescript-eslint/no-explicit-any
_log(level, msg, args) {
const time = Date.now();
const meta = { ...this.meta, ...(args ?? {}) };
(0, pretty_print_1.prettyPrint)({ level, time, msg, ...meta }, !this.isVerbose);
this.coreListeners.forEach((listener) => listener(level, msg, meta));
this.listeners.forEach((listener) => listener(level, msg, meta));
}
info(message, args) {
if (isInfoEnabled) {
this._log('info', message, args);
}
}
error(message, args) {
this._log('error', message, args);
}
debug(message, args) {
if (isDebugEnabled) {
this._log('debug', message, args);
}
}
warn(message, args) {
if (isWarnEnabled) {
this._log('warn', message, args);
}
}
// eslint-disable-next-line @typescript-eslint/no-explicit-any
log(args) {
this._log(args.level ?? 'info', args.msg, args);
}
addListener(listener) {
this.listeners.push(listener);
}
}
exports.Logger = Logger;
exports.globalLogger = new Logger();

View File

@@ -0,0 +1,4 @@
import { LockedData } from './locked-data';
export declare const createMermaidGenerator: (baseDir: string) => {
initialize: (lockedData: LockedData) => void;
};

View File

@@ -0,0 +1,203 @@
"use strict";
var __importDefault = (this && this.__importDefault) || function (mod) {
return (mod && mod.__esModule) ? mod : { "default": mod };
};
Object.defineProperty(exports, "__esModule", { value: true });
exports.createMermaidGenerator = void 0;
const fs_1 = __importDefault(require("fs"));
const path_1 = __importDefault(require("path"));
const guards_1 = require("./guards");
// Pure function to ensure diagrams directory exists
const ensureDiagramsDirectory = (diagramsDir) => {
if (!fs_1.default.existsSync(diagramsDir)) {
fs_1.default.mkdirSync(diagramsDir, { recursive: true });
}
};
// Pure function to get node ID
const getNodeId = (step, baseDir) => {
// Get relative path from the base directory
const relativePath = path_1.default.relative(baseDir, step.filePath);
// Remove common file extensions
const pathWithoutExtension = relativePath.replace(/\.(ts|js|tsx|jsx)$/, '');
// Replace slashes with underscores and dots with underscores
// Only keep alphanumeric characters and underscores
return pathWithoutExtension.replace(/[^a-zA-Z0-9]/g, '_');
};
// Pure function to get node label
const getNodeLabel = (step) => {
// Get display name for node
const displayName = step.config.name || path_1.default.basename(step.filePath, path_1.default.extname(step.filePath));
// Add node type prefix to help distinguish types
let prefix = '';
if ((0, guards_1.isApiStep)(step))
prefix = '🌐 ';
else if ((0, guards_1.isEventStep)(step))
prefix = '⚡ ';
else if ((0, guards_1.isCronStep)(step))
prefix = '⏰ ';
else if ((0, guards_1.isNoopStep)(step))
prefix = '⚙️ ';
// Create a node label with the step name
return `["${prefix}${displayName}"]`;
};
// Pure function to get node style
const getNodeStyle = (step) => {
// Apply style class based on step type
if ((0, guards_1.isApiStep)(step))
return ':::apiStyle';
if ((0, guards_1.isEventStep)(step))
return ':::eventStyle';
if ((0, guards_1.isCronStep)(step))
return ':::cronStyle';
if ((0, guards_1.isNoopStep)(step))
return ':::noopStyle';
return '';
};
// Pure function to generate connections
const generateConnections = (emits, sourceStep, steps, sourceId, baseDir) => {
const connections = [];
if (!emits || !Array.isArray(emits) || emits.length === 0) {
return '';
}
// Helper function to check if a step subscribes to a topic
const stepSubscribesToTopic = (step, topic) => {
// Event steps use regular subscribes
if ((0, guards_1.isEventStep)(step) &&
step.config.subscribes &&
Array.isArray(step.config.subscribes) &&
step.config.subscribes.includes(topic)) {
return true;
}
// Noop and API steps use virtualSubscribes
if (((0, guards_1.isNoopStep)(step) || (0, guards_1.isApiStep)(step)) &&
step.config.virtualSubscribes &&
Array.isArray(step.config.virtualSubscribes) &&
step.config.virtualSubscribes.includes(topic)) {
return true;
}
return false;
};
emits.forEach((emit) => {
const topic = typeof emit === 'string' ? emit : emit.topic;
const label = typeof emit === 'string' ? topic : emit.label || topic;
steps.forEach((targetStep) => {
if (stepSubscribesToTopic(targetStep, topic)) {
const targetId = getNodeId(targetStep, baseDir);
connections.push(` ${sourceId} -->|${label}| ${targetId}`);
}
});
});
return connections.join('\n');
};
// Pure function to generate flow diagram
const generateFlowDiagram = (flowName, steps, baseDir) => {
// Start mermaid flowchart with top-down direction
let diagram = `flowchart TD\n`;
// Add class definitions for styling with explicit text color
const classDefinitions = [
` classDef apiStyle fill:#f96,stroke:#333,stroke-width:2px,color:#fff`,
` classDef eventStyle fill:#69f,stroke:#333,stroke-width:2px,color:#fff`,
` classDef cronStyle fill:#9c6,stroke:#333,stroke-width:2px,color:#fff`,
` classDef noopStyle fill:#3f3a50,stroke:#333,stroke-width:2px,color:#fff`,
];
diagram += classDefinitions.join('\n') + '\n';
// Check if we have any steps
if (!steps || steps.length === 0) {
return diagram + ' empty[No steps in this flow]';
}
// Create node definitions with proper format
steps.forEach((step) => {
const nodeId = getNodeId(step, baseDir);
const nodeLabel = getNodeLabel(step);
const nodeStyle = getNodeStyle(step);
diagram += ` ${nodeId}${nodeLabel}${nodeStyle}\n`;
});
// Create connections between nodes
let connectionsStr = '';
steps.forEach((sourceStep) => {
const sourceId = getNodeId(sourceStep, baseDir);
// Helper function to process emissions if they exist
function processEmissions(emissionsArray, stepSource, stepsCollection, sourceIdentifier) {
if (emissionsArray && Array.isArray(emissionsArray)) {
return generateConnections(emissionsArray, stepSource, stepsCollection, sourceIdentifier, baseDir);
}
return '';
}
// Semantic variables to clarify which step types support which emission types
const supportsEmits = (0, guards_1.isApiStep)(sourceStep) || (0, guards_1.isEventStep)(sourceStep) || (0, guards_1.isCronStep)(sourceStep);
const supportsVirtualEmits = supportsEmits || (0, guards_1.isNoopStep)(sourceStep);
// Process regular emissions if supported
if (supportsEmits) {
const emitConnections = processEmissions(sourceStep.config.emits, sourceStep, steps, sourceId);
if (emitConnections) {
connectionsStr += emitConnections + '\n';
}
}
// Process virtual emissions if supported
if (supportsVirtualEmits) {
const virtualEmitConnections = processEmissions(sourceStep.config.virtualEmits, sourceStep, steps, sourceId);
if (virtualEmitConnections) {
connectionsStr += virtualEmitConnections + '\n';
}
}
});
// Add connections to the diagram
diagram += connectionsStr;
return diagram;
};
// Function to save a diagram to a file
const saveDiagram = (diagramsDir, flowName, diagram) => {
const filePath = path_1.default.join(diagramsDir, `${flowName}.mmd`);
fs_1.default.writeFileSync(filePath, diagram);
};
// Function to remove a diagram file
const removeDiagram = (diagramsDir, flowName) => {
const filePath = path_1.default.join(diagramsDir, `${flowName}.mmd`);
if (fs_1.default.existsSync(filePath)) {
fs_1.default.unlinkSync(filePath);
}
};
// Function to generate and save a diagram
const generateAndSaveDiagram = (diagramsDir, flowName, flow, baseDir) => {
const diagram = generateFlowDiagram(flowName, flow.steps, baseDir);
saveDiagram(diagramsDir, flowName, diagram);
};
// Main exported function that creates the mermaid generator
const createMermaidGenerator = (baseDir) => {
const diagramsDir = path_1.default.join(baseDir, '.mermaid');
ensureDiagramsDirectory(diagramsDir);
// Event handlers
const handleFlowCreated = (flowName, flow) => {
generateAndSaveDiagram(diagramsDir, flowName, flow, baseDir);
};
const handleFlowUpdated = (flowName, flow) => {
generateAndSaveDiagram(diagramsDir, flowName, flow, baseDir);
};
const handleFlowRemoved = (flowName) => {
removeDiagram(diagramsDir, flowName);
};
// Initialize function to hook into LockedData events
const initialize = (lockedData) => {
// Hook into flow events
lockedData.on('flow-created', (flowName) => {
handleFlowCreated(flowName, lockedData.flows[flowName]);
});
lockedData.on('flow-updated', (flowName) => {
handleFlowUpdated(flowName, lockedData.flows[flowName]);
});
lockedData.on('flow-removed', (flowName) => {
handleFlowRemoved(flowName);
});
// Generate diagrams for all existing flows
if (lockedData.flows && typeof lockedData.flows === 'object') {
Object.entries(lockedData.flows).forEach(([flowName, flow]) => {
generateAndSaveDiagram(diagramsDir, flowName, flow, baseDir);
});
}
};
// Return the public API
return {
initialize,
};
};
exports.createMermaidGenerator = createMermaidGenerator;

View File

@@ -0,0 +1,13 @@
import { Printer } from './printer';
import { TracerFactory } from './observability';
import { EventManager, InternalStateManager } from './types';
import { LockedData } from './locked-data';
import { LoggerFactory } from './logger-factory';
export type Motia = {
loggerFactory: LoggerFactory;
eventManager: EventManager;
state: InternalStateManager;
lockedData: LockedData;
printer: Printer;
tracerFactory: TracerFactory;
};

View File

@@ -0,0 +1,2 @@
"use strict";
Object.defineProperty(exports, "__esModule", { value: true });

View File

@@ -0,0 +1 @@
export {};

View File

@@ -0,0 +1,57 @@
"use strict";
var __importDefault = (this && this.__importDefault) || function (mod) {
return (mod && mod.__esModule) ? mod : { "default": mod };
};
Object.defineProperty(exports, "__esModule", { value: true });
const path_1 = __importDefault(require("path"));
const zod_to_json_schema_1 = __importDefault(require("zod-to-json-schema"));
// Add ts-node registration before dynamic imports
// eslint-disable-next-line @typescript-eslint/no-require-imports
require('ts-node').register({
transpileOnly: true,
compilerOptions: { module: 'commonjs' },
});
function isZodSchema(value) {
return Boolean(value && typeof value.safeParse === 'function' && value._def);
}
async function getConfig(filePath) {
try {
// eslint-disable-next-line @typescript-eslint/no-require-imports
const module = require(path_1.default.resolve(filePath));
// Check if the specified function exists in the module
if (!module.config) {
throw new Error(`Config not found in module ${filePath}`);
}
if (isZodSchema(module.config.input)) {
module.config.input = (0, zod_to_json_schema_1.default)(module.config.input);
}
else if (isZodSchema(module.config.bodySchema)) {
module.config.bodySchema = (0, zod_to_json_schema_1.default)(module.config.bodySchema);
}
if (module.config.responseSchema) {
for (const [status, schema] of Object.entries(module.config.responseSchema)) {
if (isZodSchema(schema)) {
module.config.responseSchema[status] = (0, zod_to_json_schema_1.default)(schema);
}
}
}
if (isZodSchema(module.config.schema)) {
module.config.schema = (0, zod_to_json_schema_1.default)(module.config.schema);
}
process.send?.(module.config);
process.exit(0);
}
catch (error) {
console.error('Error running TypeScript module:', error);
process.exit(1);
}
}
const [, , filePath] = process.argv;
if (!filePath) {
console.error('Usage: node get-config.js <file-path>');
process.exit(1);
}
getConfig(filePath).catch((err) => {
console.error('Error:', err);
process.exit(1);
});

View File

@@ -0,0 +1,12 @@
import { RpcSender } from './rpc';
export declare class Logger {
private readonly traceId;
private readonly flows;
private readonly sender;
constructor(traceId: string, flows: string[], sender: RpcSender);
private _log;
info(message: string, args?: Record<string, unknown>): void;
error(message: string, args?: Record<string, unknown>): void;
debug(message: string, args?: Record<string, unknown>): void;
warn(message: string, args?: Record<string, unknown>): void;
}

View File

@@ -0,0 +1,43 @@
"use strict";
Object.defineProperty(exports, "__esModule", { value: true });
exports.Logger = void 0;
class Logger {
constructor(traceId, flows, sender) {
this.traceId = traceId;
this.flows = flows;
this.sender = sender;
}
_log(level, message, args) {
const argsCopy = args ? { ...args } : {};
if (argsCopy.error && argsCopy.error instanceof Error) {
argsCopy.error = {
...argsCopy.error,
message: argsCopy.error.message,
stack: argsCopy.error.stack,
name: argsCopy.error.name,
};
}
const logEntry = {
...argsCopy,
level,
time: Date.now(),
traceId: this.traceId,
flows: this.flows,
msg: message,
};
this.sender.sendNoWait('log', logEntry);
}
info(message, args) {
this._log('info', message, args);
}
error(message, args) {
this._log('error', message, args);
}
debug(message, args) {
this._log('debug', message, args);
}
warn(message, args) {
this._log('warn', message, args);
}
}
exports.Logger = Logger;

View File

@@ -0,0 +1 @@
export declare const composeMiddleware: (...middlewares: any[]) => (req: any, ctx: any, handler: () => Promise<any>) => Promise<any>;

View File

@@ -0,0 +1,12 @@
"use strict";
Object.defineProperty(exports, "__esModule", { value: true });
exports.composeMiddleware = void 0;
/* eslint-disable @typescript-eslint/no-explicit-any */
const composeMiddleware = (...middlewares) => {
return async (req, ctx, handler) => {
const composedHandler = middlewares.reduceRight((nextHandler, middleware) => () => middleware(req, ctx, nextHandler), handler);
return composedHandler();
};
};
exports.composeMiddleware = composeMiddleware;
/* eslint-enable @typescript-eslint/no-explicit-any */

View File

@@ -0,0 +1 @@
export {};

View File

@@ -0,0 +1,87 @@
"use strict";
var __importDefault = (this && this.__importDefault) || function (mod) {
return (mod && mod.__esModule) ? mod : { "default": mod };
};
Object.defineProperty(exports, "__esModule", { value: true });
const path_1 = __importDefault(require("path"));
const logger_1 = require("./logger");
const middleware_compose_1 = require("./middleware-compose");
const rpc_1 = require("./rpc");
const rpc_state_manager_1 = require("./rpc-state-manager");
// eslint-disable-next-line @typescript-eslint/no-require-imports
require('dotenv').config();
// Add ts-node registration before dynamic imports
// eslint-disable-next-line @typescript-eslint/no-require-imports
require('ts-node').register({
transpileOnly: true,
compilerOptions: { module: 'commonjs' },
});
function parseArgs(arg) {
try {
return JSON.parse(arg);
}
catch {
return arg;
}
}
async function runTypescriptModule(filePath, event) {
const sender = new rpc_1.RpcSender(process);
try {
// eslint-disable-next-line @typescript-eslint/no-require-imports
const module = require(path_1.default.resolve(filePath));
// Check if the specified function exists in the module
if (typeof module.handler !== 'function') {
throw new Error(`Function handler not found in module ${filePath}`);
}
const { traceId, flows, contextInFirstArg } = event;
const logger = new logger_1.Logger(traceId, flows, sender);
const state = new rpc_state_manager_1.RpcStateManager(sender);
const emit = async (data) => sender.send('emit', data);
const streamsConfig = event.streams;
const streams = (streamsConfig ?? []).reduce((acc, streams) => {
acc[streams.name] = {
get: (groupId, id) => sender.send(`streams.${streams.name}.get`, { groupId, id }),
set: (groupId, id, data) => sender.send(`streams.${streams.name}.set`, { groupId, id, data }),
delete: (groupId, id) => sender.send(`streams.${streams.name}.delete`, { groupId, id }),
getGroup: (groupId) => sender.send(`streams.${streams.name}.getGroup`, { groupId }),
send: (channel, event) => sender.send(`streams.${streams.name}.send`, { channel, event }),
};
return acc;
}, {});
const context = { traceId, flows, logger, state, emit, streams };
sender.init();
const middlewares = Array.isArray(module.config.middleware) ? module.config.middleware : [];
const composedMiddleware = (0, middleware_compose_1.composeMiddleware)(...middlewares);
const handlerFn = () => {
return contextInFirstArg ? module.handler(context) : module.handler(event.data, context);
};
const result = await composedMiddleware(event.data, context, handlerFn);
await sender.send('result', result);
await sender.close();
process.exit(0);
// eslint-disable-next-line @typescript-eslint/no-explicit-any
}
catch (err) {
const stack = err.stack?.split('\n') ?? [];
if (stack) {
const index = stack.findIndex((line) => line.includes('src/node/node-runner'));
stack.splice(index, stack.length - index);
stack.splice(0, 1); // remove first line which has the error message
}
const error = {
message: err.message || '',
code: err.code || null,
stack: stack.join('\n'),
};
sender.sendNoWait('close', error);
}
}
const [, , filePath, arg] = process.argv;
if (!filePath) {
console.error('Usage: node nodeRunner.js <file-path> <arg>');
process.exit(1);
}
runTypescriptModule(filePath, parseArgs(arg)).catch((err) => {
console.error('Error:', err);
process.exit(1);
});

View File

@@ -0,0 +1,11 @@
import { InternalStateManager } from '../types';
import { RpcSender } from './rpc';
export declare class RpcStateManager implements InternalStateManager {
private readonly sender;
constructor(sender: RpcSender);
get<T>(traceId: string, key: string): Promise<T>;
set<T>(traceId: string, key: string, value: T): Promise<T>;
delete<T>(traceId: string, key: string): Promise<T | null>;
clear(traceId: string): Promise<void>;
getGroup<T>(groupId: string): Promise<T[]>;
}

View File

@@ -0,0 +1,24 @@
"use strict";
Object.defineProperty(exports, "__esModule", { value: true });
exports.RpcStateManager = void 0;
class RpcStateManager {
constructor(sender) {
this.sender = sender;
}
async get(traceId, key) {
return this.sender.send('state.get', { traceId, key });
}
async set(traceId, key, value) {
return this.sender.send('state.set', { traceId, key, value });
}
async delete(traceId, key) {
return this.sender.send('state.delete', { traceId, key });
}
async clear(traceId) {
await this.sender.send('state.clear', { traceId });
}
async getGroup(groupId) {
return this.sender.send('state.getGroup', { groupId });
}
}
exports.RpcStateManager = RpcStateManager;

View File

@@ -0,0 +1,9 @@
export declare class RpcSender {
private readonly process;
private readonly pendingRequests;
constructor(process: NodeJS.Process);
close(): Promise<void>;
send<T>(method: string, args: unknown): Promise<T>;
sendNoWait(method: string, args: unknown): void;
init(): void;
}

View File

@@ -0,0 +1,50 @@
"use strict";
var __importDefault = (this && this.__importDefault) || function (mod) {
return (mod && mod.__esModule) ? mod : { "default": mod };
};
Object.defineProperty(exports, "__esModule", { value: true });
exports.RpcSender = void 0;
/// <reference types="node" />
const crypto_1 = __importDefault(require("crypto"));
class RpcSender {
constructor(process) {
this.process = process;
this.pendingRequests = {};
}
async close() {
const outstandingRequests = Object.values(this.pendingRequests);
if (outstandingRequests.length > 0) {
console.error('Process ended while there are some promises outstanding');
this.process.exit(1);
}
}
send(method, args) {
return new Promise((resolve, reject) => {
const id = crypto_1.default.randomUUID();
this.pendingRequests[id] = { resolve, reject, method, args };
this.process.send?.({ type: 'rpc_request', id, method, args });
});
}
sendNoWait(method, args) {
this.process.send?.({ type: 'rpc_request', method, args });
}
init() {
this.process.on('message', (msg) => {
if (msg.type === 'rpc_response') {
const { id, result, error } = msg;
const callbacks = this.pendingRequests[id];
if (!callbacks) {
return;
}
else if (error) {
callbacks.reject(error);
}
else {
callbacks.resolve(result);
}
delete this.pendingRequests[id];
}
});
}
}
exports.RpcSender = RpcSender;

View File

@@ -0,0 +1,3 @@
import { Step } from '../types';
import { Trace, TraceGroup } from './types';
export declare const createTrace: (traceGroup: TraceGroup, step: Step) => Trace;

View File

@@ -0,0 +1,22 @@
"use strict";
Object.defineProperty(exports, "__esModule", { value: true });
exports.createTrace = void 0;
const crypto_1 = require("crypto");
const createTrace = (traceGroup, step) => {
const id = (0, crypto_1.randomUUID)();
const trace = {
id,
name: step.config.name,
correlationId: traceGroup.correlationId,
parentTraceId: traceGroup.id,
status: 'running',
startTime: Date.now(),
endTime: undefined,
entryPoint: { type: step.config.type, stepName: step.config.name },
events: [],
};
traceGroup.metadata.totalSteps++;
traceGroup.metadata.activeSteps++;
return trace;
};
exports.createTrace = createTrace;

View File

@@ -0,0 +1,14 @@
import { Logger } from '../logger';
import { Step } from '../types';
import { StateOperation, StreamOperation, TraceError } from './types';
export interface TracerFactory {
createTracer(traceId: string, step: Step, logger: Logger): Promise<Tracer> | Tracer;
clear(): Promise<void>;
}
export interface Tracer {
end(err?: TraceError): void;
stateOperation(operation: StateOperation, input: unknown): void;
emitOperation(topic: string, data: unknown, success: boolean): void;
streamOperation(streamName: string, operation: StreamOperation, input: unknown): void;
child(step: Step, logger: Logger): Tracer;
}

View File

@@ -0,0 +1,2 @@
"use strict";
Object.defineProperty(exports, "__esModule", { value: true });

View File

@@ -0,0 +1,9 @@
import { Tracer } from '.';
export declare class NoTracer implements Tracer {
end(): void;
stateOperation(): void;
emitOperation(): void;
streamOperation(): void;
clear(): void;
child(): this;
}

View File

@@ -0,0 +1,14 @@
"use strict";
Object.defineProperty(exports, "__esModule", { value: true });
exports.NoTracer = void 0;
class NoTracer {
end() { }
stateOperation() { }
emitOperation() { }
streamOperation() { }
clear() { }
child() {
return this;
}
}
exports.NoTracer = NoTracer;

View File

@@ -0,0 +1,21 @@
import { Tracer } from '.';
import { Logger } from '../logger';
import { Step } from '../types';
import { TraceManager } from './trace-manager';
import { StateOperation, StreamOperation, Trace, TraceError, TraceGroup } from './types';
export declare class StreamTracer implements Tracer {
private readonly manager;
private readonly traceGroup;
private readonly trace;
constructor(manager: TraceManager, traceGroup: TraceGroup, trace: Trace, logger: Logger);
end(err?: TraceError): void;
stateOperation(operation: StateOperation, input: unknown): void;
emitOperation(topic: string, data: unknown, success: boolean): void;
streamOperation(streamName: string, operation: StreamOperation, input: {
groupId: string;
id: string;
data?: unknown;
}): void;
child(step: Step, logger: Logger): StreamTracer;
private addEvent;
}

View File

@@ -0,0 +1,97 @@
"use strict";
Object.defineProperty(exports, "__esModule", { value: true });
exports.StreamTracer = void 0;
const create_trace_1 = require("./create-trace");
class StreamTracer {
constructor(manager, traceGroup, trace, logger) {
this.manager = manager;
this.traceGroup = traceGroup;
this.trace = trace;
logger.addListener((level, msg, args) => {
this.addEvent({
type: 'log',
timestamp: Date.now(),
level,
message: msg,
metadata: args,
});
});
}
end(err) {
if (this.trace.endTime) {
// avoiding updating twice
return;
}
this.trace.status = err ? 'failed' : 'completed';
this.trace.endTime = Date.now();
this.trace.error = err;
this.traceGroup.metadata.completedSteps++;
this.traceGroup.metadata.activeSteps--;
if (this.traceGroup.metadata.activeSteps === 0) {
if (this.traceGroup.status === 'running') {
this.traceGroup.status = 'completed';
}
this.traceGroup.endTime = Date.now();
}
if (err) {
this.traceGroup.status = 'failed';
}
this.manager.updateTrace();
this.manager.updateTraceGroup();
}
stateOperation(operation, input) {
this.addEvent({
type: 'state',
timestamp: Date.now(),
operation,
data: input,
});
}
emitOperation(topic, data, success) {
this.addEvent({
type: 'emit',
timestamp: Date.now(),
topic,
success,
data,
});
}
streamOperation(streamName, operation, input) {
if (operation === 'set') {
const lastEvent = this.trace.events[this.trace.events.length - 1];
if (lastEvent &&
lastEvent.type === 'stream' &&
lastEvent.streamName === streamName &&
lastEvent.data.groupId === input.groupId &&
lastEvent.data.id === input.id) {
lastEvent.calls++;
lastEvent.data.data = input.data;
lastEvent.maxTimestamp = Date.now();
this.traceGroup.lastActivity = lastEvent.maxTimestamp;
this.manager.updateTrace();
this.manager.updateTraceGroup();
return;
}
}
this.addEvent({
type: 'stream',
timestamp: Date.now(),
operation,
data: input,
streamName,
calls: 1,
});
}
child(step, logger) {
const trace = (0, create_trace_1.createTrace)(this.traceGroup, step);
const manager = this.manager.child(trace);
return new StreamTracer(manager, this.traceGroup, trace, logger);
}
addEvent(event) {
this.trace.events.push(event);
this.traceGroup.lastActivity = event.timestamp;
this.manager.updateTrace();
this.manager.updateTraceGroup();
}
}
exports.StreamTracer = StreamTracer;

View File

@@ -0,0 +1,12 @@
import { MotiaStream } from '../types-stream';
import { Trace, TraceGroup } from './types';
export declare class TraceManager {
private readonly traceStream;
private readonly traceGroupStream;
private readonly traceGroup;
private readonly trace;
constructor(traceStream: MotiaStream<Trace>, traceGroupStream: MotiaStream<TraceGroup>, traceGroup: TraceGroup, trace: Trace);
updateTrace(): void;
updateTraceGroup(): void;
child(trace: Trace): TraceManager;
}

View File

@@ -0,0 +1,23 @@
"use strict";
Object.defineProperty(exports, "__esModule", { value: true });
exports.TraceManager = void 0;
class TraceManager {
constructor(traceStream, traceGroupStream, traceGroup, trace) {
this.traceStream = traceStream;
this.traceGroupStream = traceGroupStream;
this.traceGroup = traceGroup;
this.trace = trace;
this.updateTrace();
this.updateTraceGroup();
}
updateTrace() {
this.traceStream.set(this.traceGroup.id, this.trace.id, this.trace);
}
updateTraceGroup() {
this.traceGroupStream.set('default', this.traceGroup.id, this.traceGroup);
}
child(trace) {
return new TraceManager(this.traceStream, this.traceGroupStream, this.traceGroup, trace);
}
}
exports.TraceManager = TraceManager;

View File

@@ -0,0 +1,11 @@
import { FileStreamAdapter } from '../streams/adapters/file-stream-adapter';
import { BaseStreamItem } from '../types-stream';
export declare class TraceStreamAdapter<TData> extends FileStreamAdapter<TData> {
private state;
private isDirty;
constructor(filePath: string, streamName: string, streamAdapter: 'file' | 'memory');
get(groupId: string, id: string): Promise<BaseStreamItem<TData> | null>;
set(groupId: string, id: string, data: TData): Promise<BaseStreamItem<TData>>;
delete(groupId: string, id: string): Promise<BaseStreamItem<TData> | null>;
getGroup(groupId: string): Promise<BaseStreamItem<TData>[]>;
}

View File

@@ -0,0 +1,53 @@
"use strict";
Object.defineProperty(exports, "__esModule", { value: true });
exports.TraceStreamAdapter = void 0;
const file_stream_adapter_1 = require("../streams/adapters/file-stream-adapter");
class TraceStreamAdapter extends file_stream_adapter_1.FileStreamAdapter {
constructor(filePath, streamName, streamAdapter) {
super(filePath ?? '', streamName);
this.state = {};
this.isDirty = false;
if (streamAdapter === 'file') {
const state = this._readFile();
Object.entries(state).forEach(([key, value]) => {
if (typeof value === 'string') {
this.state[key] = JSON.parse(value);
}
else {
this.state[key] = value;
}
});
setInterval(() => {
if (this.isDirty) {
this._writeFile(this.state);
this.isDirty = false;
}
}, 30000);
}
}
async get(groupId, id) {
const key = this._makeKey(groupId, id);
return this.state[key] ? this.state[key] : null;
}
async set(groupId, id, data) {
const key = this._makeKey(groupId, id);
this.state[key] = data;
this.isDirty = true;
return { ...data, id };
}
async delete(groupId, id) {
const key = this._makeKey(groupId, id);
const value = await this.get(groupId, id);
if (value) {
delete this.state[key];
this.isDirty = true;
}
return value;
}
async getGroup(groupId) {
return Object.entries(this.state)
.filter(([key]) => key.startsWith(groupId))
.map(([, value]) => value);
}
}
exports.TraceStreamAdapter = TraceStreamAdapter;

View File

@@ -0,0 +1,17 @@
import { TracerFactory } from '.';
import { LockedData } from '../locked-data';
import { Logger } from '../logger';
import { Step } from '../types';
import { MotiaStream } from '../types-stream';
import { StreamTracer } from './stream-tracer';
import { Trace, TraceGroup } from './types';
export declare class BaseTracerFactory implements TracerFactory {
private readonly traceStream;
private readonly traceGroupStream;
constructor(traceStream: MotiaStream<Trace>, traceGroupStream: MotiaStream<TraceGroup>);
private getAllGroups;
private deleteGroup;
clear(): Promise<void>;
createTracer(traceId: string, step: Step, logger: Logger): Promise<StreamTracer>;
}
export declare const createTracerFactory: (lockedData: LockedData) => TracerFactory;

View File

@@ -0,0 +1,86 @@
"use strict";
Object.defineProperty(exports, "__esModule", { value: true });
exports.createTracerFactory = exports.BaseTracerFactory = void 0;
const create_trace_1 = require("./create-trace");
const stream_tracer_1 = require("./stream-tracer");
const trace_manager_1 = require("./trace-manager");
const trace_stream_adapter_1 = require("./trace-stream-adapter");
const MAX_TRACE_GROUPS = process.env.MOTIA_MAX_TRACE_GROUPS //
? parseInt(process.env.MOTIA_MAX_TRACE_GROUPS)
: 50;
class BaseTracerFactory {
constructor(traceStream, traceGroupStream) {
this.traceStream = traceStream;
this.traceGroupStream = traceGroupStream;
}
async getAllGroups() {
return await this.traceGroupStream.getGroup('default');
}
async deleteGroup(group) {
const traces = await this.traceStream.getGroup(group.id);
for (const trace of traces) {
await this.traceStream.delete(group.id, trace.id);
}
await this.traceGroupStream.delete('default', group.id);
}
async clear() {
const groups = await this.getAllGroups();
for (const group of groups) {
await this.deleteGroup(group);
}
}
async createTracer(traceId, step, logger) {
const traceGroup = {
id: traceId,
name: step.config.name,
lastActivity: Date.now(),
metadata: {
completedSteps: 0,
activeSteps: 0,
totalSteps: 0,
},
correlationId: undefined,
status: 'running',
startTime: Date.now(),
};
const groups = await this.getAllGroups();
if (groups.length >= MAX_TRACE_GROUPS) {
const groupsToDelete = groups
.sort((a, b) => a.startTime - b.startTime) // date ascending
.slice(0, groups.length - MAX_TRACE_GROUPS + 1);
for (const group of groupsToDelete) {
await this.deleteGroup(group);
}
}
const trace = (0, create_trace_1.createTrace)(traceGroup, step);
const manager = new trace_manager_1.TraceManager(this.traceStream, this.traceGroupStream, traceGroup, trace);
return new stream_tracer_1.StreamTracer(manager, traceGroup, trace, logger);
}
}
exports.BaseTracerFactory = BaseTracerFactory;
const createTracerFactory = (lockedData) => {
const traceStreamName = 'motia-trace';
const traceStreamAdapter = new trace_stream_adapter_1.TraceStreamAdapter(lockedData.baseDir, traceStreamName, lockedData.streamAdapter);
const traceStream = lockedData.createStream({
filePath: traceStreamName,
hidden: true,
config: {
name: traceStreamName,
baseConfig: { storageType: 'custom', factory: () => traceStreamAdapter },
schema: null,
},
})();
const traceGroupName = 'motia-trace-group';
const traceGroupStreamAdapter = new trace_stream_adapter_1.TraceStreamAdapter(lockedData.baseDir, traceGroupName, lockedData.streamAdapter);
const traceGroupStream = lockedData.createStream({
filePath: traceGroupName,
hidden: true,
config: {
name: traceGroupName,
baseConfig: { storageType: 'custom', factory: () => traceGroupStreamAdapter },
schema: null,
},
})();
return new BaseTracerFactory(traceStream, traceGroupStream);
};
exports.createTracerFactory = createTracerFactory;

View File

@@ -0,0 +1,74 @@
import { StepConfig } from '../types';
export interface TraceGroup {
id: string;
correlationId: string | undefined;
name: string;
status: 'running' | 'completed' | 'failed';
startTime: number;
endTime?: number;
lastActivity: number;
metadata: {
completedSteps: number;
activeSteps: number;
totalSteps: number;
};
}
export type TraceError = {
message: string;
code?: string | number;
stack?: string;
};
export interface Trace {
id: string;
name: string;
correlationId?: string;
parentTraceId?: string;
status: 'running' | 'completed' | 'failed';
startTime: number;
endTime?: number;
error?: TraceError;
entryPoint: {
type: StepConfig['type'];
stepName: string;
};
events: TraceEvent[];
}
export type TraceEvent = StateEvent | EmitEvent | StreamEvent | LogEntry;
export type StateOperation = 'get' | 'getGroup' | 'set' | 'delete' | 'clear';
export type StreamOperation = 'get' | 'getGroup' | 'set' | 'delete' | 'clear' | 'send';
export interface StateEvent {
type: 'state';
timestamp: number;
operation: 'get' | 'getGroup' | 'set' | 'delete' | 'clear';
key?: string;
duration?: number;
data: unknown;
}
export interface EmitEvent {
type: 'emit';
timestamp: number;
topic: string;
success: boolean;
data: unknown;
}
export interface StreamEvent {
type: 'stream';
timestamp: number;
operation: StreamOperation;
streamName: string;
duration?: number;
maxTimestamp?: number;
data: {
groupId: string;
id: string;
data?: unknown;
};
calls: number;
}
export interface LogEntry {
type: 'log';
timestamp: number;
level: string;
message: string;
metadata?: unknown;
}

View File

@@ -0,0 +1,2 @@
"use strict";
Object.defineProperty(exports, "__esModule", { value: true });

View File

@@ -0,0 +1 @@
export declare const prettyPrint: (json: Record<string, any>, excludeDetails?: boolean) => void;

View File

@@ -0,0 +1,66 @@
"use strict";
var __importDefault = (this && this.__importDefault) || function (mod) {
return (mod && mod.__esModule) ? mod : { "default": mod };
};
Object.defineProperty(exports, "__esModule", { value: true });
exports.prettyPrint = void 0;
const colors_1 = __importDefault(require("colors"));
const stepTag = (step) => colors_1.default.bold(colors_1.default.cyan(step));
const timestampTag = (timestamp) => colors_1.default.gray(timestamp);
const traceIdTag = (traceId) => colors_1.default.gray(traceId);
const levelTags = {
error: colors_1.default.red('[ERROR]'),
info: colors_1.default.blue('[INFO]'),
warn: colors_1.default.yellow('[WARN]'),
debug: colors_1.default.gray('[DEBUG]'),
trace: colors_1.default.gray('[TRACE]'),
};
const numericTag = (value) => colors_1.default.green(value);
const stringTag = (value) => colors_1.default.cyan(value);
const booleanTag = (value) => colors_1.default.blue(value);
const arrayBrackets = ['[', ']'].map((s) => colors_1.default.gray(s));
const objectBrackets = ['{', '}'].map((s) => colors_1.default.gray(s));
// eslint-disable-next-line @typescript-eslint/no-explicit-any
const prettyPrintObject = (obj, depth = 0, parentIsLast = false, prefix = '') => {
const tab = prefix + (depth === 0 ? '' : parentIsLast ? '│ ' : '│ ');
if (depth > 2) {
return `${tab}${colors_1.default.gray('[...]')}`;
}
const entries = Object.entries(obj);
return entries
.map(([key, value], index) => {
const isLast = index === entries.length - 1;
const isObject = typeof value === 'object' && value !== null;
const prefix = isLast && !isObject ? '└' : '├';
if (isObject) {
const subObject = prettyPrintObject(value, depth + 1, isLast, tab);
const [start, end] = Array.isArray(value) ? arrayBrackets : objectBrackets;
return `${tab}${prefix} ${key}: ${start}\n${subObject}\n${tab}${isLast ? '└' : '│'} ${end}`;
}
let printedValue = value;
if (typeof value === 'number') {
printedValue = numericTag(String(value));
}
else if (typeof value === 'boolean') {
printedValue = booleanTag(String(value));
}
else if (typeof value === 'string') {
printedValue = stringTag(value);
}
return `${tab}${prefix} ${key}: ${printedValue}`;
})
.join('\n');
};
// eslint-disable-next-line @typescript-eslint/no-explicit-any
const prettyPrint = (json, excludeDetails = false) => {
// eslint-disable-next-line @typescript-eslint/no-unused-vars
const { time, traceId, msg, flows, level, step, ...details } = json;
const levelTag = levelTags[level];
const timestamp = timestampTag(`[${new Date(time).toLocaleTimeString()}]`);
const objectHasKeys = Object.keys(details).length > 0;
console.log(`${timestamp} ${traceIdTag(traceId)} ${levelTag} ${stepTag(step)} ${msg}`);
if (objectHasKeys && !excludeDetails) {
console.log(prettyPrintObject(details));
}
};
exports.prettyPrint = prettyPrint;

View File

@@ -0,0 +1,46 @@
import { ValidationError } from './step-validator';
import { Step } from './types';
import { Stream } from './types-stream';
export declare class Printer {
private readonly baseDir;
constructor(baseDir: string);
stepTag: string;
flowTag: string;
created: string;
building: string;
built: string;
updated: string;
removed: string;
printInvalidEmit(step: Step, emit: string): void;
printStepCreated(step: Step): void;
printStepUpdated(step: Step): void;
printStepRemoved(step: Step): void;
printFlowCreated(flowName: string): void;
printFlowUpdated(flowName: string): void;
printFlowRemoved(flowName: string): void;
printStreamCreated(stream: Stream): void;
printStreamUpdated(stream: Stream): void;
printStreamRemoved(stream: Stream): void;
printInvalidEmitConfiguration(step: Step, emit: string): void;
printInvalidSchema(topic: string, step: Step[]): void;
printValidationError(stepPath: string, validationError: ValidationError): void;
getRelativePath(filePath: string): string;
getStepType(step: Step): string;
getStepPath(step: Step): string;
getStreamPath(stream: Stream): string;
}
export declare class NoPrinter extends Printer {
constructor();
printInvalidEmit(): void;
printStepCreated(): void;
printStepUpdated(): void;
printStepRemoved(): void;
printFlowCreated(): void;
printFlowUpdated(): void;
printFlowRemoved(): void;
printStepType(): void;
printStepPath(): void;
printStreamCreated(): void;
printStreamUpdated(): void;
printStreamRemoved(): void;
}

View File

@@ -0,0 +1,125 @@
"use strict";
var __importDefault = (this && this.__importDefault) || function (mod) {
return (mod && mod.__esModule) ? mod : { "default": mod };
};
Object.defineProperty(exports, "__esModule", { value: true });
exports.NoPrinter = exports.Printer = void 0;
const colors_1 = __importDefault(require("colors"));
const path_1 = __importDefault(require("path"));
const guards_1 = require("./guards");
const stepTag = colors_1.default.bold(colors_1.default.magenta('Step'));
const flowTag = colors_1.default.bold(colors_1.default.blue('Flow'));
const streamTag = colors_1.default.bold(colors_1.default.green('Stream'));
const created = colors_1.default.green('➜ [CREATED]');
const building = colors_1.default.yellow('⚡ [BUILDING]');
const built = colors_1.default.green('✓ [BUILT]');
const updated = colors_1.default.yellow('➜ [UPDATED]');
const removed = colors_1.default.red('➜ [REMOVED]');
const invalidEmit = colors_1.default.red('➜ [INVALID EMIT]');
const error = colors_1.default.red('[ERROR]');
const warning = colors_1.default.yellow('[WARNING]');
class Printer {
constructor(baseDir) {
this.baseDir = baseDir;
this.stepTag = stepTag;
this.flowTag = flowTag;
this.created = created;
this.building = building;
this.built = built;
this.updated = updated;
this.removed = removed;
}
printInvalidEmit(step, emit) {
console.log(`${invalidEmit} ${stepTag} ${this.getStepType(step)} ${this.getStepPath(step)} tried to emit an event not defined in the step config: ${colors_1.default.yellow(emit)}`);
}
printStepCreated(step) {
console.log(`${created} ${stepTag} ${this.getStepType(step)} ${this.getStepPath(step)} created`);
}
printStepUpdated(step) {
console.log(`${updated} ${stepTag} ${this.getStepType(step)} ${this.getStepPath(step)} updated`);
}
printStepRemoved(step) {
console.log(`${removed} ${stepTag} ${this.getStepType(step)} ${this.getStepPath(step)} removed`);
}
printFlowCreated(flowName) {
console.log(`${created} ${flowTag} ${colors_1.default.bold(colors_1.default.cyan(flowName))} created`);
}
printFlowUpdated(flowName) {
console.log(`${updated} ${flowTag} ${colors_1.default.bold(colors_1.default.cyan(flowName))} updated`);
}
printFlowRemoved(flowName) {
console.log(`${removed} ${flowTag} ${colors_1.default.bold(colors_1.default.cyan(flowName))} removed`);
}
printStreamCreated(stream) {
console.log(`${created} ${streamTag} ${this.getStreamPath(stream)} created`);
}
printStreamUpdated(stream) {
console.log(`${updated} ${streamTag} ${this.getStreamPath(stream)} updated`);
}
printStreamRemoved(stream) {
console.log(`${removed} ${streamTag} ${this.getStreamPath(stream)} removed`);
}
printInvalidEmitConfiguration(step, emit) {
console.log(`${warning} ${stepTag} ${this.getStepType(step)} ${this.getStepPath(step)} emits to ${colors_1.default.yellow(emit)}, but there is no subscriber defined`);
}
printInvalidSchema(topic, step) {
console.log(`${error} Topic ${colors_1.default.bold(colors_1.default.blue(topic))} has incompatible schemas in the following steps:`);
step.forEach((step) => {
console.log(`${colors_1.default.red(' ✖')} ${this.getStepPath(step)}`);
});
}
printValidationError(stepPath, validationError) {
const relativePath = this.getRelativePath(stepPath);
console.log(`${error} ${colors_1.default.bold(colors_1.default.cyan(relativePath))}`);
validationError.errors?.forEach((error) => {
if (error.path) {
console.log(`${colors_1.default.red('│')} ${colors_1.default.yellow(`${error.path}`)}: ${error.message}`);
}
else {
console.log(`${colors_1.default.red('│')} ${colors_1.default.yellow('✖')} ${error.message}`);
}
});
console.log(`${colors_1.default.red('└─')} ${colors_1.default.red(validationError.error)} `);
}
getRelativePath(filePath) {
return path_1.default.relative(this.baseDir, filePath);
}
getStepType(step) {
if ((0, guards_1.isApiStep)(step))
return colors_1.default.gray('(API)');
if ((0, guards_1.isEventStep)(step))
return colors_1.default.gray('(Event)');
if ((0, guards_1.isCronStep)(step))
return colors_1.default.gray('(Cron)');
if ((0, guards_1.isNoopStep)(step))
return colors_1.default.gray('(Noop)');
return colors_1.default.gray('(Unknown)');
}
getStepPath(step) {
const stepPath = this.getRelativePath(step.filePath);
return colors_1.default.bold(colors_1.default.cyan(stepPath));
}
getStreamPath(stream) {
const streamPath = this.getRelativePath(stream.filePath);
return colors_1.default.bold(colors_1.default.magenta(streamPath));
}
}
exports.Printer = Printer;
class NoPrinter extends Printer {
constructor() {
super('');
}
printInvalidEmit() { }
printStepCreated() { }
printStepUpdated() { }
printStepRemoved() { }
printFlowCreated() { }
printFlowUpdated() { }
printFlowRemoved() { }
printStepType() { }
printStepPath() { }
printStreamCreated() { }
printStreamUpdated() { }
printStreamRemoved() { }
}
exports.NoPrinter = NoPrinter;

View File

@@ -0,0 +1,7 @@
import { SpawnOptions } from 'child_process';
export type CommunicationType = 'rpc' | 'ipc';
export interface CommunicationConfig {
type: CommunicationType;
spawnOptions: SpawnOptions;
}
export declare function createCommunicationConfig(command: string, projectRoot?: string): CommunicationConfig;

View File

@@ -0,0 +1,18 @@
"use strict";
Object.defineProperty(exports, "__esModule", { value: true });
exports.createCommunicationConfig = createCommunicationConfig;
function createCommunicationConfig(command, projectRoot) {
const type = command === 'python' && process.platform === 'win32' ? 'rpc' : 'ipc';
const spawnOptions = {
stdio: type === 'rpc'
? ['pipe', 'pipe', 'inherit'] // RPC: capture stdout
: ['inherit', 'inherit', 'inherit', 'ipc'], // IPC: include IPC channel
};
if (command === 'python') {
spawnOptions.env = {
...process.env,
PYTHONPATH: projectRoot || process.cwd(),
};
}
return { type, spawnOptions };
}

View File

@@ -0,0 +1,31 @@
import { ChildProcess } from 'child_process';
import { CommunicationType } from './communication-config';
import { RpcHandler, MessageCallback } from './rpc-processor-interface';
import { Logger } from '../logger';
export interface ProcessManagerOptions {
command: string;
args: string[];
logger: Logger;
context?: string;
projectRoot?: string;
}
export declare class ProcessManager {
private options;
private child?;
private processor?;
private communicationType?;
constructor(options: ProcessManagerOptions);
spawn(): Promise<ChildProcess>;
handler<TInput, TOutput = unknown>(method: string, handler: RpcHandler<TInput, TOutput>): void;
onMessage<T = unknown>(callback: MessageCallback<T>): void;
onProcessClose(callback: (code: number | null) => void): void;
onProcessError(callback: (error: Error & {
code?: string;
}) => void): void;
onStderr(callback: (data: Buffer) => void): void;
onStdout(callback: (data: Buffer) => void): void;
kill(): void;
close(): void;
get process(): ChildProcess | undefined;
get commType(): CommunicationType | undefined;
}

View File

@@ -0,0 +1,88 @@
"use strict";
Object.defineProperty(exports, "__esModule", { value: true });
exports.ProcessManager = void 0;
const child_process_1 = require("child_process");
const communication_config_1 = require("./communication-config");
const step_handler_rpc_processor_1 = require("../step-handler-rpc-processor");
const step_handler_rpc_stdin_processor_1 = require("../step-handler-rpc-stdin-processor");
class ProcessManager {
constructor(options) {
this.options = options;
}
async spawn() {
const { command, args, logger, context = 'Process', projectRoot } = this.options;
// Get communication configuration
const commConfig = (0, communication_config_1.createCommunicationConfig)(command, projectRoot);
this.communicationType = commConfig.type;
logger.debug(`[${context}] Spawning process`, {
command,
args,
communicationType: this.communicationType,
});
// Spawn the process
this.child = (0, child_process_1.spawn)(command, args, commConfig.spawnOptions);
// Create appropriate processor based on communication type
this.processor = this.communicationType === 'rpc' ? new step_handler_rpc_stdin_processor_1.RpcStdinProcessor(this.child) : new step_handler_rpc_processor_1.RpcProcessor(this.child);
// Initialize the processor
await this.processor.init();
return this.child;
}
handler(method, handler) {
if (!this.processor) {
throw new Error('Process not spawned yet. Call spawn() first.');
}
this.processor.handler(method, handler);
}
onMessage(callback) {
if (!this.processor) {
throw new Error('Process not spawned yet. Call spawn() first.');
}
this.processor.onMessage(callback);
}
onProcessClose(callback) {
if (!this.child) {
throw new Error('Process not spawned yet. Call spawn() first.');
}
this.child.on('close', callback);
}
onProcessError(callback) {
if (!this.child) {
throw new Error('Process not spawned yet. Call spawn() first.');
}
this.child.on('error', callback);
}
onStderr(callback) {
if (!this.child) {
throw new Error('Process not spawned yet. Call spawn() first.');
}
this.child.stderr?.on('data', callback);
}
onStdout(callback) {
if (!this.child) {
throw new Error('Process not spawned yet. Call spawn() first.');
}
// Only for non-RPC mode (in RPC mode, stdout is used for communication)
if (this.communicationType !== 'rpc') {
this.child.stdout?.on('data', callback);
}
}
kill() {
if (this.child) {
this.child.kill('SIGKILL');
}
}
close() {
if (this.processor) {
this.processor.close();
}
this.processor = undefined;
this.child = undefined;
}
get process() {
return this.child;
}
get commType() {
return this.communicationType;
}
}
exports.ProcessManager = ProcessManager;

View File

@@ -0,0 +1,9 @@
export type RpcHandler<TInput, TOutput> = (input: TInput) => Promise<TOutput>;
export type MessageCallback<T = unknown> = (message: T) => void;
export interface RpcProcessorInterface {
handler<TInput, TOutput = unknown>(method: string, handler: RpcHandler<TInput, TOutput>): void;
handle(method: string, input: unknown): Promise<unknown>;
onMessage<T = unknown>(callback: MessageCallback<T>): void;
init(): Promise<void>;
close(): void;
}

View File

@@ -0,0 +1,2 @@
"use strict";
Object.defineProperty(exports, "__esModule", { value: true });

View File

@@ -0,0 +1,60 @@
import sys
import json
import importlib.util
import os
import platform
def sendMessage(text):
'sends a Node IPC message to parent proccess'
# encode message as json string + newline in bytes
bytesMessage = (json.dumps(text) + "\n").encode('utf-8')
# Handle Windows differently
if platform.system() == 'Windows':
# On Windows, write to stdout
sys.stdout.buffer.write(bytesMessage)
sys.stdout.buffer.flush()
else:
# On Unix systems, use the file descriptor approach
NODEIPCFD = int(os.environ["NODE_CHANNEL_FD"])
os.write(NODEIPCFD, bytesMessage)
async def run_python_module(file_path: str) -> None:
try:
module_dir = os.path.dirname(os.path.abspath(file_path))
if module_dir not in sys.path:
sys.path.insert(0, module_dir)
flows_dir = os.path.dirname(module_dir)
if flows_dir not in sys.path:
sys.path.insert(0, flows_dir)
spec = importlib.util.spec_from_file_location("dynamic_module", file_path)
if spec is None or spec.loader is None:
raise ImportError(f"Could not load module from {file_path}")
module = importlib.util.module_from_spec(spec)
module.__package__ = os.path.basename(module_dir)
spec.loader.exec_module(module)
if not hasattr(module, 'config'):
raise AttributeError(f"No 'config' found in module {file_path}")
if 'middleware' in module.config:
del module.config['middleware']
sendMessage(module.config)
except Exception as error:
print('Error running Python module:', str(error), file=sys.stderr)
sys.exit(1)
if __name__ == "__main__":
if len(sys.argv) < 2:
sys.exit(1)
file_path = sys.argv[1]
import asyncio
asyncio.run(run_python_module(file_path))

View File

@@ -0,0 +1,35 @@
import os
import platform
from typing import Union
from motia_rpc_communication import RpcCommunication
from motia_ipc_communication import IpcCommunication
def create_communication() -> Union[RpcCommunication, IpcCommunication]:
"""
Create appropriate communication instance based on platform and environment.
Logic:
- Python + Windows = RPC (stdin/stdout)
- Other cases with NODE_CHANNEL_FD = IPC
- Fallback = RPC
"""
# Check if we're on Windows
is_windows = platform.system() == 'Windows'
# Check if IPC file descriptor is available
has_ipc_fd = "NODE_CHANNEL_FD" in os.environ
if is_windows:
# On Windows, always use RPC
return RpcCommunication()
elif has_ipc_fd:
# On Unix with IPC FD available, use IPC
try:
return IpcCommunication()
except RuntimeError:
# Fallback to RPC if IPC fails to initialize
return RpcCommunication()
else:
# Fallback to RPC
return RpcCommunication()

View File

@@ -0,0 +1,24 @@
from typing import Any, List, Optional
from motia_type_definitions import HandlerResult
from motia_rpc import RpcSender
from motia_rpc_state_manager import RpcStateManager
from motia_logger import Logger
from motia_dot_dict import DotDict
class Context:
def __init__(
self,
trace_id: str,
flows: List[str],
rpc: RpcSender,
streams: DotDict,
):
self.trace_id = trace_id
self.flows = flows
self.rpc = rpc
self.state = RpcStateManager(rpc)
self.streams = streams
self.logger = Logger(self.trace_id, self.flows, rpc)
async def emit(self, event: Any) -> Optional[HandlerResult]:
return await self.rpc.send('emit', event)

View File

@@ -0,0 +1,16 @@
class DotDict(dict):
def __getattr__(self, key):
try:
value = self[key]
return DotDict(value) if isinstance(value, dict) else value
except KeyError:
raise AttributeError(f"No such attribute: {key}")
def __setattr__(self, key, value):
self[key] = value
def __delattr__(self, key):
try:
del self[key]
except KeyError:
raise AttributeError(f"No such attribute: {key}")

View File

@@ -0,0 +1,145 @@
import uuid
import asyncio
import json
import sys
import os
from typing import Any, Dict, Optional, Callable
def serialize_for_json(obj: Any) -> Any:
"""Convert Python objects to JSON-serializable types"""
if hasattr(obj, '__dict__'):
return obj.__dict__
elif hasattr(obj, '_asdict'):
return obj._asdict()
elif isinstance(obj, (list, tuple)):
return [serialize_for_json(item) for item in obj]
elif isinstance(obj, dict):
return {k: serialize_for_json(v) for k, v in obj.items()}
else:
return obj
class IpcCommunication:
"""IPC communication using file descriptors"""
def __init__(self):
self.executing = True
self.pending_requests: Dict[str, asyncio.Future] = {}
self.ipc_reader_task: Optional[asyncio.Task] = None
self.message_handlers: Dict[str, Callable] = {}
self.ipc_fd: Optional[int] = None
# Get IPC file descriptor
if "NODE_CHANNEL_FD" in os.environ:
try:
self.ipc_fd = int(os.environ["NODE_CHANNEL_FD"])
except (ValueError, TypeError):
raise RuntimeError("Invalid NODE_CHANNEL_FD environment variable")
else:
raise RuntimeError("NODE_CHANNEL_FD environment variable not found")
def send_no_wait(self, method: str, args: Any) -> None:
"""Send IPC request without waiting for response"""
request = {
'type': 'rpc_request',
'method': method,
'args': args
}
try:
json_str = json.dumps(request, default=serialize_for_json)
message_bytes = (json_str + "\n").encode('utf-8')
os.write(self.ipc_fd, message_bytes)
except Exception as e:
print(f"ERROR: Failed to send IPC request: {e}", file=sys.stderr)
async def send(self, method: str, args: Any) -> Any:
"""Send IPC request and wait for response"""
request_id = str(uuid.uuid4())
future = asyncio.Future()
self.pending_requests[request_id] = future
request = {
'type': 'rpc_request',
'id': request_id,
'method': method,
'args': args
}
try:
json_str = json.dumps(request, default=serialize_for_json)
message_bytes = (json_str + "\n").encode('utf-8')
os.write(self.ipc_fd, message_bytes)
except Exception as e:
future.set_exception(e)
return await future
return await future
def _handle_message(self, msg: Dict[str, Any]) -> None:
"""Handle incoming message from Node.js"""
msg_type = msg.get('type')
if msg_type == 'rpc_response':
request_id = msg.get('id')
if request_id in self.pending_requests:
future = self.pending_requests[request_id]
del self.pending_requests[request_id]
error = msg.get('error')
if error is not None:
future.set_exception(Exception(str(error)))
else:
future.set_result(msg.get('result'))
elif msg_type in self.message_handlers:
try:
self.message_handlers[msg_type](msg)
except Exception as e:
print(f"ERROR: Handler for {msg_type} failed: {e}", file=sys.stderr)
async def _read_ipc(self) -> None:
"""Read messages from IPC file descriptor in background"""
loop = asyncio.get_event_loop()
buffer = ""
while self.executing:
try:
data = await loop.run_in_executor(None, os.read, self.ipc_fd, 4096)
if not data:
break
buffer += data.decode('utf-8')
lines = buffer.split('\n')
buffer = lines[-1] # Keep incomplete line in buffer
for line in lines[:-1]:
if line.strip():
try:
msg = json.loads(line.strip())
self._handle_message(msg)
except json.JSONDecodeError as e:
print(f"WARNING: Failed to parse JSON: {e}", file=sys.stderr)
except OSError:
# IPC channel closed
break
except Exception as e:
print(f"ERROR: Reading IPC failed: {e}", file=sys.stderr)
await asyncio.sleep(0.1)
async def init(self) -> None:
"""Initialize IPC communication"""
if not self.ipc_reader_task:
self.ipc_reader_task = asyncio.create_task(self._read_ipc())
def close(self) -> None:
"""Close IPC communication"""
self.executing = False
for future in self.pending_requests.values():
if not future.done():
future.set_exception(Exception("IPC connection closed"))
self.pending_requests.clear()
if self.ipc_reader_task and not self.ipc_reader_task.done():
self.ipc_reader_task.cancel()

View File

@@ -0,0 +1,40 @@
import time
from typing import Any, Dict, Optional
from motia_rpc import RpcSender
class Logger:
def __init__(self, trace_id: str, flows: list[str], rpc: RpcSender):
self.trace_id = trace_id
self.flows = flows
self.rpc = rpc
def _log(self, level: str, message: str, args: Optional[Dict[str, Any]] = None) -> None:
log_entry = {
"level": level,
"time": int(time.time() * 1000),
"traceId": self.trace_id,
"flows": self.flows,
"msg": message
}
if args:
# Use our serializer to ensure args are JSON-serializable
if hasattr(args, '__dict__'):
args = vars(args)
elif not isinstance(args, dict):
args = {"data": args}
log_entry.update(args)
self.rpc.send_no_wait('log', log_entry)
def info(self, message: str, args: Optional[Any] = None) -> None:
self._log("info", message, args)
def error(self, message: str, args: Optional[Any] = None) -> None:
self._log("error", message, args)
def debug(self, message: str, args: Optional[Any] = None) -> None:
self._log("debug", message, args)
def warn(self, message: str, args: Optional[Any] = None) -> None:
self._log("warn", message, args)

View File

@@ -0,0 +1,17 @@
from typing import Any, Callable
from functools import reduce
from motia_context import Context
def compose_middleware(*middlewares):
"""Compose multiple middleware functions into a single middleware"""
def compose_two(f: Callable, g: Callable) -> Callable:
async def composed(data: Any, context: Context, next_fn: Callable):
async def wrapped_next(d=data):
return await g(d, context, next_fn)
return await f(data, context, wrapped_next)
return composed
if not middlewares:
return lambda data, context, next_fn: next_fn()
return reduce(compose_two, middlewares)

View File

@@ -0,0 +1,39 @@
from typing import Any, Union
from motia_communication_factory import create_communication
from motia_rpc_communication import RpcCommunication
from motia_ipc_communication import IpcCommunication
def serialize_for_json(obj: Any) -> Any:
"""Convert Python objects to JSON-serializable types"""
if hasattr(obj, '__dict__'):
return obj.__dict__
elif hasattr(obj, '_asdict'): # For namedtuples
return obj._asdict()
elif isinstance(obj, (list, tuple)):
return [serialize_for_json(item) for item in obj]
elif isinstance(obj, dict):
return {k: serialize_for_json(v) for k, v in obj.items()}
else:
return obj
class RpcSender:
"""Unified communication interface that delegates to appropriate implementation"""
def __init__(self):
self._communication: Union[RpcCommunication, IpcCommunication] = create_communication()
def send_no_wait(self, method: str, args: Any) -> None:
"""Send request without waiting for response"""
return self._communication.send_no_wait(method, args)
async def send(self, method: str, args: Any) -> Any:
"""Send request and wait for response"""
return await self._communication.send(method, args)
async def init(self) -> None:
"""Initialize communication"""
return await self._communication.init()
def close(self) -> None:
"""Close communication"""
return self._communication.close()

View File

@@ -0,0 +1,125 @@
import uuid
import asyncio
import json
import sys
from typing import Any, Dict, Optional, Callable
def serialize_for_json(obj: Any) -> Any:
"""Convert Python objects to JSON-serializable types"""
if hasattr(obj, '__dict__'):
return obj.__dict__
elif hasattr(obj, '_asdict'):
return obj._asdict()
elif isinstance(obj, (list, tuple)):
return [serialize_for_json(item) for item in obj]
elif isinstance(obj, dict):
return {k: serialize_for_json(v) for k, v in obj.items()}
else:
return obj
class RpcCommunication:
"""RPC communication using stdin/stdout"""
def __init__(self):
self.executing = True
self.pending_requests: Dict[str, asyncio.Future] = {}
self.stdin_reader_task: Optional[asyncio.Task] = None
self.message_handlers: Dict[str, Callable] = {}
def send_no_wait(self, method: str, args: Any) -> None:
"""Send RPC request without waiting for response"""
request = {
'type': 'rpc_request',
'method': method,
'args': args
}
try:
json_str = json.dumps(request, default=serialize_for_json)
print(json_str, flush=True)
except Exception as e:
print(f"ERROR: Failed to send RPC request: {e}", file=sys.stderr)
async def send(self, method: str, args: Any) -> Any:
"""Send RPC request and wait for response"""
request_id = str(uuid.uuid4())
future = asyncio.Future()
self.pending_requests[request_id] = future
request = {
'type': 'rpc_request',
'id': request_id,
'method': method,
'args': args
}
try:
json_str = json.dumps(request, default=serialize_for_json)
print(json_str, flush=True)
except Exception as e:
future.set_exception(e)
return await future
return await future
def _handle_message(self, msg: Dict[str, Any]) -> None:
"""Handle incoming message from Node.js"""
msg_type = msg.get('type')
if msg_type == 'rpc_response':
request_id = msg.get('id')
if request_id in self.pending_requests:
future = self.pending_requests[request_id]
del self.pending_requests[request_id]
error = msg.get('error')
if error is not None:
future.set_exception(Exception(str(error)))
else:
future.set_result(msg.get('result'))
elif msg_type in self.message_handlers:
try:
self.message_handlers[msg_type](msg)
except Exception as e:
print(f"ERROR: Handler for {msg_type} failed: {e}", file=sys.stderr)
async def _read_stdin(self) -> None:
"""Read messages from stdin in background"""
loop = asyncio.get_event_loop()
while self.executing:
try:
line = await loop.run_in_executor(None, sys.stdin.readline)
if not line:
break
line = line.strip()
if line:
try:
msg = json.loads(line)
self._handle_message(msg)
except json.JSONDecodeError as e:
print(f"WARNING: Failed to parse JSON: {e}", file=sys.stderr)
except Exception as e:
print(f"ERROR: Reading stdin failed: {e}", file=sys.stderr)
await asyncio.sleep(0.1)
async def init(self) -> None:
"""Initialize RPC communication"""
if not self.stdin_reader_task:
self.stdin_reader_task = asyncio.create_task(self._read_stdin())
def close(self) -> None:
"""Close RPC communication"""
self.executing = False
for future in self.pending_requests.values():
if not future.done():
future.set_exception(Exception("RPC connection closed"))
self.pending_requests.clear()
if self.stdin_reader_task and not self.stdin_reader_task.done():
self.stdin_reader_task.cancel()

View File

@@ -0,0 +1,70 @@
import asyncio
import functools
import sys
from typing import Any
from motia_rpc import RpcSender
class RpcStateManager:
def __init__(self, rpc: RpcSender):
self.rpc = rpc
self._loop = asyncio.get_event_loop()
async def get(self, trace_id: str, key: str) -> asyncio.Future[Any]:
result = await self.rpc.send('state.get', {'traceId': trace_id, 'key': key})
if result is None:
return {'data': None}
elif isinstance(result, dict):
if 'data' not in result:
return {'data': result}
return result
async def get_group(self, group_id: str) -> asyncio.Future[Any]:
result = await self.rpc.send('state.getGroup', {'groupId': group_id})
if result is None:
return {'data': None}
elif isinstance(result, dict):
if 'data' not in result:
return {'data': result}
return result
async def getGroup(self, trace_id: str, key: str) -> asyncio.Future[Any]:
return await self.get_group(trace_id, key)
async def set(self, trace_id: str, key: str, value: Any) -> asyncio.Future[None]:
future = await self.rpc.send('state.set', {'traceId': trace_id, 'key': key, 'value': value})
return future
async def delete(self, trace_id: str, key: str) -> asyncio.Future[None]:
return await self.rpc.send('state.delete', {'traceId': trace_id, 'key': key})
async def clear(self, trace_id: str) -> asyncio.Future[None]:
return await self.rpc.send('state.clear', {'traceId': trace_id})
# Add wrappers to handle non-awaited coroutines
def __getattribute__(self, name):
attr = super().__getattribute__(name)
if name in ('get', 'set', 'delete', 'clear') and asyncio.iscoroutinefunction(attr):
@functools.wraps(attr)
def wrapper(*args, **kwargs):
coro = attr(*args, **kwargs)
# Check if this is being awaited
frame = sys._getframe(1)
if frame.f_code.co_name != '__await__':
# Not being awaited, schedule in background
# But we need to make sure this task completes before the handler returns
# So we'll return the task for the caller to await if needed
task = asyncio.create_task(coro)
# Add error handling for the background task
def handle_exception(t):
if t.done() and not t.cancelled() and t.exception():
print(f"Unhandled exception in background task: {t.exception()}", file=sys.stderr)
task.add_done_callback(handle_exception)
return task
# Being awaited, return coroutine as normal
return coro
return wrapper
return attr

View File

@@ -0,0 +1,56 @@
import asyncio
import functools
import sys
from typing import Any, Dict
from motia_rpc import RpcSender
class RpcStreamManager:
def __init__(self, stream_name: str,rpc: RpcSender):
self.rpc = rpc
self.stream_name = stream_name
self._loop = asyncio.get_event_loop()
async def get(self, group_id: str, id: str) -> asyncio.Future[Any]:
result = await self.rpc.send(f'streams.{self.stream_name}.get', {'groupId': group_id, 'id': id})
return result
async def set(self, group_id: str, id: str, data: Any) -> asyncio.Future[None]:
future = await self.rpc.send(f'streams.{self.stream_name}.set', {'groupId': group_id, 'id': id, 'data': data})
return future
async def delete(self, group_id: str, id: str) -> asyncio.Future[None]:
return await self.rpc.send(f'streams.{self.stream_name}.delete', {'groupId': group_id, 'id': id})
async def getGroup(self, group_id: str) -> asyncio.Future[None]:
return await self.rpc.send(f'streams.{self.stream_name}.getGroup', {'groupId': group_id})
async def get_group(self, group_id: str) -> asyncio.Future[None]:
return await self.getGroup(group_id)
async def send(self, channel: Dict, event: Dict) -> asyncio.Future[None]:
return await self.rpc.send(f'streams.{self.stream_name}.send', {'channel': channel, 'event': event})
# Add wrappers to handle non-awaited coroutines
def __getattribute__(self, name):
attr = super().__getattribute__(name)
if name in ('get', 'set', 'delete', 'getGroup') and asyncio.iscoroutinefunction(attr):
@functools.wraps(attr)
def wrapper(*args, **kwargs):
coro = attr(*args, **kwargs)
# Check if this is being awaited
frame = sys._getframe(1)
if frame.f_code.co_name != '__await__':
# Not being awaited, schedule in background
# But we need to make sure this task completes before the handler returns
# So we'll return the task for the caller to await if needed
task = asyncio.create_task(coro)
# Add error handling for the background task
def handle_exception(t):
if t.done() and not t.cancelled() and t.exception():
print(f"Unhandled exception in background task: {t.exception()}", file=sys.stderr)
task.add_done_callback(handle_exception)
return task
# Being awaited, return coroutine as normal
return coro
return wrapper
return attr

View File

@@ -0,0 +1,33 @@
from typing import TypeVar, Callable, Coroutine, Union, Dict, List, Optional
from types import SimpleNamespace
# Generic type for JSON schema validation
T = TypeVar('T')
# Custom types
JsonSchema = Dict[str, any]
ValidationResult = Dict[str, Union[bool, Dict, str]]
HandlerResult = Optional[Dict[str, any]]
HandlerFunction = Callable[..., Coroutine[any, any, HandlerResult]]
class FlowConfig(SimpleNamespace):
type: str
input: Optional[JsonSchema]
bodySchema: Optional[JsonSchema]
class HandlerArgs(SimpleNamespace):
traceId: str
flows: List[str]
data: Union[Dict, SimpleNamespace]
contextInFirstArg: bool
class ApiResponse:
def __init__(
self,
status: int,
body: Dict[str, any],
headers: Optional[Dict[str, str]] = None
):
self.status = status
self.body = body
self.headers = headers or {'Content-Type': 'application/json'}

View File

@@ -0,0 +1,108 @@
import sys
import json
import importlib.util
import os
import asyncio
import traceback
from typing import Callable, List, Dict
from motia_rpc import RpcSender
from motia_context import Context
from motia_middleware import compose_middleware
from motia_rpc_stream_manager import RpcStreamManager
from motia_dot_dict import DotDict
def parse_args(arg: str) -> Dict:
"""Parse command line arguments into HandlerArgs"""
try:
return json.loads(arg)
except json.JSONDecodeError:
print('Error parsing args:', arg)
return arg
async def run_python_module(file_path: str, rpc: RpcSender, args: Dict) -> None:
"""Execute a Python module with the given arguments"""
try:
module_dir = os.path.dirname(os.path.abspath(file_path))
flows_dir = os.path.dirname(module_dir)
for path in [module_dir, flows_dir]:
if path not in sys.path:
sys.path.insert(0, path)
spec = importlib.util.spec_from_file_location("dynamic_module", file_path)
if spec is None or spec.loader is None:
raise ImportError(f"Could not load module from {file_path}")
module = importlib.util.module_from_spec(spec)
module.__package__ = os.path.basename(module_dir)
spec.loader.exec_module(module)
if not hasattr(module, "handler"):
raise AttributeError(f"Function 'handler' not found in module {file_path}")
config = module.config
trace_id = args.get("traceId")
flows = args.get("flows") or []
data = args.get("data")
context_in_first_arg = args.get("contextInFirstArg")
streams_config = args.get("streams") or []
streams = DotDict()
for item in streams_config:
name = item.get("name")
streams[name] = RpcStreamManager(name, rpc)
context = Context(trace_id, flows, rpc, streams)
middlewares: List[Callable] = config.get("middleware", [])
composed_middleware = compose_middleware(*middlewares)
async def handler_fn():
if context_in_first_arg:
return await module.handler(context)
else:
return await module.handler(data, context)
result = await composed_middleware(data, context, handler_fn)
if result:
await rpc.send('result', result)
rpc.send_no_wait("close", None)
rpc.close()
except Exception as error:
stack_list = traceback.format_exception(type(error), error, error.__traceback__)
# We're removing the first two and last item
# 0: Traceback (most recent call last):
# 1: File "python-runner.py", line 82, in run_python_module
# 2: File "python-runner.py", line 69, in run_python_module
# -1: Exception: message
stack_list = stack_list[3:-1]
rpc.send_no_wait("close", {
"message": str(error),
"stack": "\n".join(stack_list)
})
rpc.close()
if __name__ == "__main__":
if len(sys.argv) < 2:
print("Usage: python pythonRunner.py <file-path> <arg>", file=sys.stderr)
sys.exit(1)
file_path = sys.argv[1]
arg = sys.argv[2] if len(sys.argv) > 2 else None
rpc = RpcSender()
try:
loop = asyncio.get_running_loop()
except RuntimeError:
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
args = parse_args(arg) if arg else None
tasks = asyncio.gather(rpc.init(), run_python_module(file_path, rpc, args))
loop.run_until_complete(tasks)

Some files were not shown because too many files have changed in this diff Show More