feat: Actor.charge()

.gitignore

@@ -27,3 +27,5 @@ apify_storage
 crawlee_storage
 storage
 .turbo
+*.tgz
+mise.toml

package.json

@@ -40,7 +40,10 @@
         "build": "turbo run build",
         "ci:build": "turbo run build --cache-dir=\".turbo\"",
         "test": "vitest run --silent",
-        "test:e2e": "node test/e2e/run.mjs",
+        "test:e2e": "npm run test:e2e:scrapers && npm run test:e2e:sdk",
+        "test:e2e:scrapers": "node test/e2e/runScraperTests.mjs",
+        "test:e2e:sdk": "npm run test:e2e:sdk:tarball && node test/e2e/runSdkTests.mjs",
+        "test:e2e:sdk:tarball": "cd packages/apify; mv $(npm pack | tail -n 1) ../../test/e2e/apify.tgz",
         "coverage": "vitest --coverage",
         "release": "npm run build && lerna publish from-package --contents dist",
         "publish:next": "lerna publish --canary --preid beta --dist-tag next",

packages/apify/package.json

@@ -62,7 +62,7 @@
         "@crawlee/core": "^3.9.0",
         "@crawlee/types": "^3.9.0",
         "@crawlee/utils": "^3.9.0",
-        "apify-client": "^2.10.0",
+        "apify-client": "^2.11.1",
         "fs-extra": "^11.2.0",
         "ow": "^0.28.2",
         "semver": "^7.5.4",

packages/apify/src/actor.ts

@@ -37,6 +37,8 @@ import {
 import ow from 'ow';
 
 import { Configuration } from './configuration';
+import { ChargingManager } from './internals/charging';
+import type { ChargeOptions, ChargeResult } from './internals/charging';
 import { KeyValueStore } from './key_value_store';
 import { PlatformEventManager } from './platform_event_manager';
 import type { ProxyConfigurationOptions } from './proxy_configuration';
@@ -86,11 +88,14 @@ export class Actor<Data extends Dictionary = Dictionary> {
      */
     private isRebooting = false;
 
+    private chargingManager: ChargingManager;
+
     constructor(options: ConfigurationOptions = {}) {
         // use default configuration object if nothing overridden (it fallbacks to env vars)
         this.config = Object.keys(options).length === 0 ? Configuration.getGlobalConfig() : new Configuration(options);
         this.apifyClient = this.newClient();
         this.eventManager = new PlatformEventManager(this.config);
+        this.chargingManager = new ChargingManager(this.config, this.apifyClient);
     }
 
     /**
@@ -222,6 +227,9 @@ export class Actor<Data extends Dictionary = Dictionary> {
         log.debug(`Default storages purged`);
 
         Configuration.storage.enterWith(this.config);
+
+        await this.chargingManager.init();
+        log.debug(`ChargingManager initialized`);
     }
 
     /**
@@ -609,13 +617,29 @@ export class Actor<Data extends Dictionary = Dictionary> {
      *
      * @param item Object or array of objects containing data to be stored in the default dataset.
      * The objects must be serializable to JSON and the JSON representation of each object must be smaller than 9MB.
+     * @param eventName If provided, the method will attempt to charge for the event for each pushed item.
      * @ignore
      */
-    async pushData(item: Data | Data[]): Promise<void> {
+    async pushData(item: Data | Data[]): Promise<void>;
+    async pushData(item: Data | Data[], eventName: string): Promise<ChargeResult>;
+    async pushData(item: Data | Data[], eventName?: string | undefined): Promise<ChargeResult | void> {
         this._ensureActorInit('pushData');
 
         const dataset = await this.openDataset();
-        return dataset.pushData(item);
+
+        const maxChargedCount = eventName !== undefined ? this.chargingManager.calculateMaxEventChargeCountWithinLimit(eventName) : Infinity;
+        const toCharge = Array.isArray(item) ? item.length : 1;
+
+        if (toCharge > maxChargedCount) {
+            const items = Array.isArray(item) ? item : [item];
+            await dataset.pushData(items.slice(0, toCharge));
+        } else {
+            await dataset.pushData(item);
+        }
+
+        if (eventName) {
+            return await this.chargingManager.charge({ eventName, count: Math.min(toCharge, maxChargedCount) });
+        }
     }
 
     /**
@@ -896,6 +920,32 @@ export class Actor<Data extends Dictionary = Dictionary> {
         return undefined;
     }
 
+    /**
+     * Charge for a specified number of events - sub-operations of the Actor.
+     *
+     * @param options The name of the event to charge for and the number of events to be charged.
+     */
+    async charge(options: ChargeOptions): Promise<ChargeResult> {
+        this._ensureActorInit('charge');
+        return this.chargingManager.charge(options);
+    }
+
+    /**
+     * Get the maximum amount of money that the Actor is allowed to charge.
+     */
+    getMaxTotalChargeUsd(): number {
+        this._ensureActorInit('getMaxTotalChargeUsd');
+        return this.chargingManager.getMaxTotalChargeUsd();
+    }
+
+    /**
+     * Get the number of events with given name that the Actor has charged for so far.
+     */
+    getChargedEventCount(eventName: string): number {
+        this._ensureActorInit('getChargedEventCount');
+        return this.chargingManager.getChargedEventCount(eventName);
+    }
+
     /**
      * Modifies Actor env vars so parsing respects the structure of {@apilink ApifyEnv} interface.
      */
@@ -1304,9 +1354,15 @@ export class Actor<Data extends Dictionary = Dictionary> {
      *
      * @param item Object or array of objects containing data to be stored in the default dataset.
      * The objects must be serializable to JSON and the JSON representation of each object must be smaller than 9MB.
+     * @param eventName If provided, the method will attempt to charge for the event for each pushed item.
      */
-    static async pushData<Data extends Dictionary = Dictionary>(item: Data | Data[]): Promise<void> {
-        return Actor.getDefaultInstance().pushData(item);
+    static async pushData<Data extends Dictionary = Dictionary>(item: Data | Data[]): Promise<void>;
+    static async pushData<Data extends Dictionary = Dictionary>(item: Data | Data[], eventName: string): Promise<ChargeResult>;
+    static async pushData<Data extends Dictionary = Dictionary>(item: Data | Data[], eventName?: string): Promise<ChargeResult | void> {
+        if (eventName === undefined) {
+            return await Actor.getDefaultInstance().pushData(item);
+        }
+        return await Actor.getDefaultInstance().pushData(item, eventName);
     }
 
     /**
@@ -1512,6 +1568,29 @@ export class Actor<Data extends Dictionary = Dictionary> {
         return Actor.getDefaultInstance().createProxyConfiguration(proxyConfigurationOptions);
     }
 
+    /**
+     * Charge for a specified number of events - sub-operations of the Actor.
+     *
+     * @param options The name of the event to charge for and the number of events to be charged.
+     */
+    static async charge(options: ChargeOptions): Promise<ChargeResult> {
+        return Actor.getDefaultInstance().charge(options);
+    }
+
+    /**
+     * Get the maximum amount of money that the Actor is allowed to charge.
+     */
+    static getMaxTotalChargeUsd(): number {
+        return Actor.getDefaultInstance().getMaxTotalChargeUsd();
+    }
+
+    /**
+     * Get the number of events with given name that the Actor has charged for so far.
+     */
+    static getChargedEventCount(eventName: string): number {
+        return Actor.getDefaultInstance().getChargedEventCount(eventName);
+    }
+
     /**
      * Returns a new {@apilink ApifyEnv} object which contains information parsed from all the Apify environment variables.
      *

packages/apify/src/configuration.ts

@@ -31,6 +31,7 @@ export interface ConfigurationOptions extends CoreConfigurationOptions {
     userId?: string;
     inputSecretsPrivateKeyPassphrase?: string;
     inputSecretsPrivateKeyFile?: string;
+    maxTotalChargeUsd?: number;
     metaOrigin?: typeof META_ORIGINS[keyof typeof META_ORIGINS];
 }
 
@@ -164,6 +165,7 @@ export class Configuration extends CoreConfiguration {
         ACTOR_TASK_ID: 'actorTaskId',
         ACTOR_WEB_SERVER_PORT: 'containerPort',
         ACTOR_WEB_SERVER_URL: 'containerUrl',
+        ACTOR_MAX_TOTAL_CHARGE_USD: 'maxTotalChargeUsd',
     };
 
     protected static override INTEGER_VARS = [...super.INTEGER_VARS, 'proxyPort', 'containerPort', 'metamorphAfterSleepMillis'];