import { IProductsBuilderService } from '../../products/services/products.service';
import { IAppVersion, IBuilderInit, IGenerateWebhookLink, INotificationsHandler, IProcessingFailure, IActionProcessorInput, INotificationProcessorInput, IProductNotificationTemplate, IStorageProcessorInput, IRegisterWebhook, WebhookEventTypes, IJobProcessorInput, IMessageBrokerPublishInput, IMessageBrokerSubscribeInput, StepEventTypes, IActionRequest, IActionSamples, IStepEvent, IProductAppEnvs, IProductEnv, IRetryMeta, IStorageRequest, IProductStorageEnvs, HttpMethods, ILogData, LogEventStatus, LogEventTypes, IAppRetryPolicy, IParsedIndexes, IParsedSample, FetchRemoteCachePayload, ISessionInput, ISessionOutput, ISessionPayload, ISessionRefreshPayload } from '../../types';
import { IPricingResult } from '../../pricing/pricing.types';
export interface IProcessorService {
    processNotification(action: INotificationProcessorInput): Promise<any>;
    processAction(data: IActionProcessorInput): Promise<any>;
    processStorage(data: IStorageProcessorInput): Promise<any>;
    processJob(data: IJobProcessorInput): Promise<any>;
    registerWebhook(data: IRegisterWebhook): Promise<void>;
    generateWebhookLink(data: IGenerateWebhookLink): Promise<string>;
    generateSession(data: ISessionInput): Promise<any>;
}
export default class ProcessorService implements IProcessorService {
    private user_id;
    private process_id;
    private input;
    private workspace_id;
    private public_key;
    private webhookApi;
    private token;
    private productBuilderService;
    private appBuilderService;
    private processorApiService;
    private logService;
    private pricingService;
    private inputService;
    private requestTrackerService;
    private start;
    private end;
    private processingOutput;
    private processEnv;
    private component;
    private productId;
    private processingFailure;
    private doneWithProcessing;
    private productTag;
    private pricingTag;
    private apps;
    private baseLogs;
    private clone;
    private environment;
    private redisClient?;
    private published;
    private requestTime;
    private totalRequests;
    private queues;
    private _privateKey;
    private accessKey;
    private featureExecutionContext?;
    private cacheService;
    /** When a notification fails, which channel (push/email/sms/callbacks) we were in, for failure log type */
    private lastNotificationFailureType?;
    /** Reuse broker connections when multiple produce steps use the same broker (avoids ~1–3s connection setup per step). Never logged or exposed. */
    private brokerServicePool;
    /** Reuse SMTP transporter so we don't open a new connection per email (avoids multi-second handshake per send). Never logged or exposed. */
    private mailTransporterPool;
    private mailPoolKeyToSharedKey;
    /** Reuse Firebase Admin app per project (avoids re-init per push). Never logged or exposed. */
    private firebaseAppPool;
    private firebasePoolKeyToSharedKey;
    /** Reuse SMS client per config (Twilio/Nexmo/Plivo). Never logged or exposed. */
    private smsClientPool;
    private smsPoolKeyToSharedKey;
    /** Healthcheck worker interval IDs; cleared and repopulated only when the healthcheck set or config changes. */
    private healthcheckWorkerIntervals;
    /** Timer that periodically checks for healthcheck list changes (default 60s). Workers are only recreated when the fingerprint changes. */
    private healthcheckRefreshTimer;
    private static readonly HEALTHCHECK_REFRESH_MS;
    /** Fingerprint of current worker set (tag:env:interval) so we only clear/recreate when the list or config actually changed. */
    private healthcheckWorkerFingerprint;
    constructor({ workspace_id, public_key, user_id, token, env_type, private_key, access_key, redis_client, queues, preInitializedProductBuilder, }: IBuilderInit & {
        private_key: string;
        preInitializedProductBuilder?: IProductsBuilderService;
    });
    /** Correlate nested action/broker/notification records with their parent feature run. */
    setFeatureExecutionContext(context: {
        feature_id: string;
        feature_tag: string;
    }): void;
    /**
     * Pool key for broker service reuse (same broker type + config => same connection).
     */
    private getBrokerPoolKey;
    /**
     * Get or create a broker service instance so we reuse the same connection for the same broker.
     */
    private getBrokerServiceForPublish;
    /**
     * Disconnect all pooled broker connections (e.g. at end of feature run).
     * Safe to call if pool is empty or a broker has no disconnect().
     */
    disconnectBrokerConnections(): Promise<void>;
    /**
     * Pool key for mail transporter reuse (host/port/user only; no password). Same SMTP config => same connection.
     */
    private getMailTransporterPoolKey;
    /**
     * Get or create a mail transporter so we reuse the same SMTP connection instead of opening a new one per send.
     * When scope.workspaceId and scope.product are provided, uses shared registry across instances.
     */
    private getMailTransporter;
    /**
     * Close all pooled mail transporters (e.g. at end of feature run). Safe if pool is empty.
     */
    disconnectMailTransporters(): Promise<void>;
    /**
     * Get or create Firebase Admin app by project_id so we reuse the same app per project (Expo client is already module-level singleton).
     * When scope.workspaceId and scope.product are provided, uses shared registry across instances.
     */
    private getFirebaseApp;
    /**
     * Close all pooled Firebase apps (e.g. at end of feature run). Safe if pool is empty.
     */
    disconnectFirebaseApps(): Promise<void>;
    /**
     * Pool key for SMS client (provider + identifier + sender; no secrets). Same config => reuse client.
     */
    private getSmsClientPoolKey;
    /**
     * Get or create SMS client so we reuse Twilio/Nexmo/Plivo client per config.
     * When scope.workspaceId and scope.product are provided, uses shared registry across instances.
     */
    private getSmsClient;
    /**
     * Clear pooled SMS clients (e.g. at end of feature run). Safe if pool is empty.
     */
    disconnectSmsClients(): Promise<void>;
    /**
     * Pre-warm broker connections for the given product/env and broker tags.
     * Call this before step execution so the first produce step does not pay connection setup cost.
     * Security: same auth as runBrokerPublish (product + env from caller). Secrets resolved in-memory
     * only; no config/URLs/credentials logged or persisted. Pool is per-processor-instance (one run).
     */
    warmBrokerConnections(options: {
        product: string;
        env: string;
        brokerTags: string[];
    }): Promise<void>;
    /**
     * Run a single healthcheck (one env) by dispatching to the correct processor based on probe type.
     * Supports app, database, feature, graph, storage, and events. Falls back to processAction when probe is missing (app/event only).
     * Public so the SDK monitor worker can call it with full healthcheck (including probe).
     */
    runOneHealthcheck(healthcheck: {
        tag: string;
        app?: string;
        event?: string;
        retries?: number;
        probe?: {
            type?: string;
            app?: string;
            database?: string;
            feature?: string;
            workflow?: string;
            graph?: string;
            events?: string;
            messageBroker?: string;
            message_broker?: string;
            storage?: string;
            cache?: string;
            vector?: string;
            notification?: string;
            channels?: string[];
            event?: string;
            input?: any;
        };
    }, envSlug: string, decryptedInput: any, productTag: string): Promise<any>;
    /**
     * Build a stable fingerprint for the current healthcheck list so we only recreate workers when the set or config changes.
     */
    private healthcheckFingerprint;
    /**
     * Start healthcheck workers for all products/environments after Redis is connected.
     * This is called automatically in the constructor if redisClient is present.
     * Dispatches to app, database, feature, graph, storage, or events processor based on probe type.
     * Every HEALTHCHECK_REFRESH_MS we re-fetch the list and only clear/recreate workers when the fingerprint
     * (tag:env:interval set) has changed, so existing timers are not reset when nothing changed.
     */
    private startHealthcheckWorkers;
    /**
     * Manually trigger healthcheck processing for all healthchecks (can be called externally if needed).
     * Dispatches to app, database, feature, graph, storage, or events processor based on probe type.
     */
    processAllHealthchecksForProduct(productTag: string): Promise<void>;
    generateSession(payload: ISessionInput): Promise<ISessionOutput>;
    refreshSession(payload: ISessionRefreshPayload): Promise<ISessionOutput>;
    decryptSession(data: ISessionPayload): Promise<any>;
    registerWebhook(data: IRegisterWebhook): Promise<void>;
    generateWebhookLink(data: IGenerateWebhookLink): Promise<string>;
    intializeProduct(additional_logs: Partial<ILogData>): Promise<void>;
    initializePricing(additional_logs: Partial<ILogData>, access_tag: string): Promise<void>;
    fetchEnv(env: string, additional_logs: Partial<ILogData>): Promise<IProductEnv>;
    constructJSONDataPayloads(object: Partial<IActionRequest>, additional_logs: Partial<ILogData>, samples: IActionSamples, event: IStepEvent, loopIndex?: number): Promise<{}>;
    generatePayload(obj: Record<string, unknown>, event: IStepEvent | null, additional_logs: Partial<ILogData>, sample?: IParsedSample[], index?: IParsedIndexes, loopIndex?: number): Promise<Record<string, unknown>>;
    generateStringValues(value: string, app: string, additional_logs: Partial<ILogData>, sample?: Array<IParsedSample>, index?: IParsedIndexes, key?: string, loopIndex?: number): Promise<any>;
    generateOperatorValues(value: string, app: string, additional_logs: Partial<ILogData>, sample?: Array<IParsedSample>, index?: IParsedIndexes, key?: string, loopIndex?: number): Promise<any>;
    sumValues(value: string, app: string, additional_logs: Partial<ILogData>, sample?: Array<IParsedSample>, index?: IParsedIndexes, key?: string, loopIndex?: number): Promise<any>;
    subtractValues(value: string, app: string, additional_logs: Partial<ILogData>, sample?: Array<IParsedSample>, index?: IParsedIndexes, key?: string, loopIndex?: number): Promise<any>;
    concatValues(value: string, app: string, additional_logs: Partial<ILogData>, sample?: Array<IParsedSample>, index?: IParsedIndexes, key?: string, loopIndex?: number): Promise<any>;
    uppercaseValue(value: string, app: string, additional_logs: Partial<ILogData>, sample?: Array<IParsedSample>, index?: IParsedIndexes, key?: string, loopIndex?: number): Promise<string>;
    lowercaseValue(value: string, app: string, additional_logs: Partial<ILogData>, sample?: Array<IParsedSample>, index?: IParsedIndexes, key?: string, loopIndex?: number): Promise<string>;
    dateFormatValue(value: string, app: string, additional_logs: Partial<ILogData>, sample?: Array<IParsedSample>, index?: IParsedIndexes, key?: string, loopIndex?: number): Promise<string>;
    replaceValue(value: string, app: string, additional_logs: Partial<ILogData>, sample?: Array<IParsedSample>, index?: IParsedIndexes, key?: string, loopIndex?: number): Promise<string>;
    substringValues(value: string, app: string, additional_logs: Partial<ILogData>, sample?: Array<IParsedSample>, index?: IParsedIndexes, key?: string, loopIndex?: number): Promise<any>;
    trimValues(value: string, app: string, additional_logs: Partial<ILogData>, sample?: Array<IParsedSample>, index?: IParsedIndexes, key?: string, loopIndex?: number): Promise<any>;
    pickValue(value: string, app: string, additional_logs: Partial<ILogData>, sample?: Array<IParsedSample>, index?: IParsedIndexes, key?: string, loopIndex?: number): Promise<any>;
    filterValue(value: string, app: string, additional_logs: Partial<ILogData>, sample?: Array<IParsedSample>, index?: IParsedIndexes, key?: string, loopIndex?: number): Promise<any[]>;
    findValue(value: string, app: string, additional_logs: Partial<ILogData>, sample?: Array<IParsedSample>, index?: IParsedIndexes, key?: string, loopIndex?: number): Promise<any>;
    splitValues(value: string, app: string, additional_logs: Partial<ILogData>, sample?: Array<IParsedSample>, index?: IParsedIndexes, key?: string, loopIndex?: number): Promise<any>;
    joinArrays(value: string, app: string, additional_logs: Partial<ILogData>, sample?: Array<IParsedSample>, index?: IParsedIndexes, key?: string, loopIndex?: number): Promise<any>;
    generateInputValue(input: Record<string, unknown>, stages: Array<string>): Promise<any>;
    generateSequenceValue(stages: Array<string>, indexLocator?: number, indexValue?: number): Promise<any>;
    fetchOutputValueAfterStrippingLocators(stages: Array<string>, output: Record<string, unknown>, indexLocator?: number, indexValue?: number, stageIndex?: number): any;
    generateDefaultValue(sample: IParsedSample[], index: IParsedIndexes): Promise<string | number | true>;
    generateVariableValue(stages: Array<string>): Promise<unknown>;
    generateConstantValue(stages: Array<string>): Promise<string>;
    decorateValue(value: string, sample: IParsedSample): string;
    generateAuthValue(stages: Array<string>, app: string, sample: IParsedSample[], additional_logs: Partial<ILogData>): Promise<string>;
    fetchAuthData(app_tag: string, additional_logs: Partial<ILogData>): Promise<Record<string, unknown>>;
    processEvent(event: IStepEvent): Promise<any>;
    runJobs(job: any, additional_logs?: Partial<ILogData>): Promise<any>;
    /**
     * Record a failed phase on the job execution tracker (e.g. from BullMQ 'failed' handler).
     * Call when the worker job fails so the tracker is updated even if runJobs' catch didn't run.
     */
    recordJobExecutionPhaseFailed(jobId: string, errorMessage?: string, errorCode?: string): Promise<void>;
    /**
     * Get job data from Redis
     */
    private getJobData;
    /**
     * Update job status in Redis
     */
    private updateJobStatus;
    /**
     * Add job execution record to history
     */
    private addJobExecution;
    /**
     * Calculate retry delay based on job configuration
     */
    private calculateJobRetry;
    /**
     * Process database action job (predefined database actions)
     * Integrates with the database action manager to execute predefined operations
     */
    private processDatabaseAction;
    /**
     * Process database operation job (direct CRUD operations)
     * Handles operations like insert, find, update, delete, aggregate
     */
    private processDatabaseOperation;
    /**
     * Process graph action job (predefined graph actions)
     * Integrates with the graph service to execute predefined graph operations
     */
    private processGraphAction;
    /**
     * Process graph operation job (direct graph operations)
     * Handles operations like createNode, findNodes, createRelationship, traverse, etc.
     */
    private processGraphOperation;
    /**
     * Process feature job
     * Executes a feature using the feature service
     */
    private processFeature;
    getAndStoreAuth(appEnv: IProductAppEnvs, access_tag: string): Promise<string>;
    fetchThirdPartyApp(access_tag: string): Promise<IAppVersion & {
        workspace_id: string;
        _id: string;
        version: string;
    }>;
    processConditionalCheck(event: IStepEvent, additional_logs?: Partial<ILogData>): Promise<boolean>;
    extractLoopIndexes(event: IStepEvent, additional_logs?: Partial<ILogData>): Promise<number[]>;
    runAction(event: IStepEvent, additional_logs: Partial<ILogData>, returnValue?: boolean, bootstrapData?: {
        action: any;
        app_env: any;
        retries: number;
        app_active: boolean;
        recipient_workspace_id: string;
        product_env_mapping?: IProductAppEnvs | null;
    }): Promise<any>;
    processRequest(payload: IRetryMeta, event: IStepEvent, retries: IAppRetryPolicy, additional_logs: Partial<ILogData>, returnValue?: boolean): Promise<true | IProcessingFailure | {
        process_id: string;
        status: boolean;
        data: {
            status: string;
            message: string;
            data: {
                id: `${string}-${string}-${string}-${string}-${string}`;
                account_id: number;
                amount: string;
                currency: any;
                card_hash: `${string}-${string}-${string}-${string}-${string}`;
                card_pan: string;
                masked_pan: string;
                city: any;
                state: any;
                address_1: any;
                address_2: string;
                zip_code: any;
                cvv: number;
                expiration: string;
                send_to: string;
                bin_check_name: string;
                card_type: string;
                name_on_card: any;
                created_at: string;
                is_active: boolean;
                callback_url: string;
            };
        };
    } | {
        process_id: string;
        status?: undefined;
        data?: undefined;
    }>;
    processPricingCost(additional_logs: Partial<ILogData>): Promise<IPricingResult>;
    addToSuccessOutput(event: IStepEvent, output: any, additional_logs: Partial<ILogData>): Promise<void>;
    checkIsSuccessful(): boolean;
    addToFailureOutput(e: any, event: IStepEvent, payload: IRetryMeta | any, additional_logs: Partial<ILogData>, policy?: Partial<IAppRetryPolicy>): Promise<IProcessingFailure>;
    generateRetryMetrices(error_code: string, retries: IAppRetryPolicy): {
        allow_fail: boolean;
        max: number;
        retry_at: number;
    };
    sendActionRequest(base_url: string, resource: string, payload: IActionRequest, method: HttpMethods, env: string): Promise<any>;
    processStorage(action: IStorageProcessorInput): Promise<{
        process_id: string;
        output: any;
    }>;
    processMessageBrokerSubscribe(data: IMessageBrokerSubscribeInput): Promise<void | {
        process_id: string;
    }>;
    processMessageBrokerPublish(data: IMessageBrokerPublishInput): Promise<{
        process_id: string;
        output: {
            published: boolean;
        };
    }>;
    processJob(job: IJobProcessorInput, additional_logs?: Partial<ILogData>): Promise<{
        job_id: string;
        status: 'scheduled' | 'queued';
        scheduled_at: number;
        recurring: boolean;
        next_run_at?: number;
    }>;
    /** Expo: expoClient() is already a module-level singleton (one axios instance), so no per-run pool needed. */
    sendExpoNotification(payload: {
        title: string;
        body: string;
        data: string;
    }, device_tokens: Array<string>): Promise<void>;
    sendFirebaseNotification(payload: {
        title: string;
        body: string;
        data: string;
    }, device_tokens: Array<string>, credentials: any, scope?: {
        workspaceId?: string;
        product?: string;
    }): Promise<void>;
    ProcessExpoNotification(notification: INotificationsHandler, template: IProductNotificationTemplate, payload: any, additional_logs: Partial<ILogData>, logType?: LogEventTypes): Promise<void>;
    ProcessFirebaseNotification(notification: INotificationsHandler, template: IProductNotificationTemplate, payload: any, additional_logs: Partial<ILogData>, logType?: LogEventTypes, scope?: {
        workspaceId?: string;
        product?: string;
    }): Promise<void>;
    runNotification(notification: IStepEvent, additional_logs: Partial<ILogData>, bootstrapData?: {
        notification: any;
        message: any;
        env_config: any;
        product_tag?: string;
        env?: string;
        process_id?: string;
        session?: string;
        cache?: string;
    }): Promise<void>;
    runStorage(data: IStepEvent, additional_logs?: Partial<ILogData>, bootstrapData?: {
        storage: any;
        storage_env: any;
        private_key?: string;
    }): Promise<any>;
    runBrokerSubscribe(data: IStepEvent, additional_logs?: Partial<ILogData>): Promise<void>;
    runBrokerPublish(data: IStepEvent, additional_logs?: Partial<ILogData>): Promise<{
        published: boolean;
    }>;
    processStorageRequest(data: IStepEvent, input: IStorageRequest, storageEnv: IProductStorageEnvs, additional_logs: Partial<ILogData>, decryptionKey?: string): Promise<IProcessingFailure | {
        url: string;
    }>;
    writeResult(status: LogEventStatus, retryable?: boolean): Promise<void>;
    /**
     * Separate credentials into prefixed (e.g., 'headers:Authorization') and non-prefixed (e.g., 'api_key').
     * Prefixed credentials are applied directly to the correct section after resolution.
     * Non-prefixed credentials go through InputResolver to determine their placement.
     */
    private separateCredentials;
    /**
     * Check if a key exists in the action schema for a given section (headers, body, params, query).
     * Returns true if the key is defined in the schema, false otherwise.
     */
    private isKeyInActionSchema;
    /**
     * Apply prefixed credentials (e.g., 'headers:Authorization') to resolved input.
     * Credentials are applied with lower priority - existing values in resolvedInput take precedence.
     * Only applies credentials if the action schema defines the corresponding field.
     */
    private applyPrefixedCredentials;
    validateActionDataMappingInput(input: any, type: StepEventTypes | WebhookEventTypes): Promise<void>;
    processAction(action: IActionProcessorInput): Promise<any>;
    processNotification(action: INotificationProcessorInput): Promise<{
        process_id: string;
        output: {
            sent: boolean;
        };
    }>;
    /**
     * Fire-and-forget log of notification message send to integrations backend (for reprocessing).
     * Input is encrypted with the product private key before sending. Does not block or affect latency; errors are swallowed.
     */
    private fireAndForgetLogNotificationMessage;
    /**
     * Fetch notification message logs (send history) with time filters. Secured via user auth.
     * Each item's input is decrypted with the product private key when stored as an encrypted string.
     */
    getNotificationMessageLogs(options: {
        product_tag?: string;
        product_id?: string;
        env?: string;
        notification_tag?: string;
        status?: string;
        type?: string;
        process_id?: string;
        start_date?: string;
        end_date?: string;
        page?: number;
        limit?: number;
    }): Promise<{
        items: any[];
        total: number;
        page: number;
        limit: number;
        hasMore: boolean;
    }>;
    fetchRemoteCaches(payload: FetchRemoteCachePayload): Promise<import("../../types").IRemoteCache[]>;
    private getUserAccess;
    /**
     * Add data to cache - fire-and-forget pattern.
     * Does not block the main operation.
     */
    private addToCache;
    private fetchFromCache;
    /**
     * Writes the healthcheck result to Redis cache for fast status retrieval.
     */
    writeHealthcheckResultToCache(data: any, result: any): Promise<void>;
    /**
     * Fetches the latest healthcheck status for a product/env from Redis cache.
     */
    getHealthcheckStatusFromCache(productTag: string, envSlug: string): Promise<any>;
    /**
     * Updates the healthcheck in the remote DB for a product with all envs' results.
     */
    updateHealthcheckOnProcessor(productTag: string, envs: any[]): Promise<void>;
}