destinationPush to utils and type updates

Author: alexanderkirtzel

packages/destinations/node/aws/src/push.ts

@@ -6,5 +6,5 @@ export const push: PushFn = async function (event, config) {
 
   if (firehose) pushFirehose([{ event }], firehose);
 
-  return { queue: [] }; // @TODO
+  return;
 };

packages/destinations/node/aws/src/types/index.ts

@@ -1,5 +1,8 @@
 import type { DestinationNode } from '@elbwalker/source-node';
-import type { Handler } from '@elbwalker/types';
+import type {
+  Handler,
+  Destination as WalkerOSDestination,
+} from '@elbwalker/types';
 import type {
   FirehoseClient,
   FirehoseClientConfig,
@@ -10,7 +13,7 @@ export interface Destination
   init: InitFn;
 }
 
-export type PushFn = DestinationNode.PushFn<Custom, CustomEvent>;
+export type PushFn = WalkerOSDestination.PushFn<Custom, CustomEvent>;
 export type InitFn = DestinationNode.InitFn<PartialConfig, Config>;
 
 export type Config = {

packages/destinations/node/bigquery/src/push.ts

@@ -8,7 +8,7 @@ export const push: PushFn = async function (event, config) {
 
   await client.dataset(datasetId).table(tableId).insert(rows);
 
-  return { queue: [] };
+  return;
 };
 
 export const mapEvent = (event: WalkerOS.Event): Row => {

packages/destinations/node/bigquery/src/types/index.ts

@@ -1,13 +1,16 @@
 import type { DestinationNode } from '@elbwalker/source-node';
-import type { Handler } from '@elbwalker/types';
+import type {
+  Handler,
+  Destination as WalkerOSDestination,
+} from '@elbwalker/types';
 import type { BigQuery, BigQueryOptions } from '@google-cloud/bigquery';
 
 export interface Destination
   extends DestinationNode.Destination<Custom, CustomEvent> {
   init: InitFn;
 }
 
-export type PushFn = DestinationNode.PushFn<Custom, CustomEvent>;
+export type PushFn = WalkerOSDestination.PushFn<Custom, CustomEvent>;
 export type InitFn = DestinationNode.InitFn<PartialConfig, Config>;
 
 export type Config = {

packages/destinations/node/etag/src/push.ts

@@ -7,7 +7,7 @@ import { getParameters } from '@elbwalker/destination-core-etag';
 export const push: PushFn = async function (pushEvent, config) {
   const { custom } = config;
 
-  if (!custom) return {};
+  if (!custom) return;
 
   // @TODO
   let pageTitle;
@@ -33,8 +33,6 @@ export const push: PushFn = async function (pushEvent, config) {
       );
     }),
   );
-
-  return { queue: [] }; // @TODO
 };
 
 async function sendRequest(custom: Custom, path: string, body?: string) {

packages/destinations/node/etag/src/types/index.ts

@@ -1,5 +1,6 @@
 import type { DestinationNode } from '@elbwalker/source-node';
 import type { DestinationCoreEtag } from '@elbwalker/destination-core-etag';
+import type { Destination as WalkerOSDestination } from '@elbwalker/types';
 
 export interface Destination
   extends DestinationNode.Destination<Custom, CustomEvent> {}
@@ -11,7 +12,7 @@ export type PartialConfig = DestinationNode.Config<
   Partial<CustomEvent>
 >;
 
-export type PushFn = DestinationNode.PushFn<Custom, CustomEvent>;
+export type PushFn = WalkerOSDestination.PushFn<Custom, CustomEvent>;
 
 export interface Custom extends DestinationCoreEtag.Config {}
 

packages/destinations/node/meta/src/push.ts

@@ -34,7 +34,7 @@ export const push: PushFn = async function (event, config, mapping) {
 
   return eventRequest.execute().then(
     () => {
-      return {};
+      return;
     },
     (err: unknown) => {
       throw err;

packages/destinations/node/meta/src/types/index.ts

@@ -1,12 +1,16 @@
 import type { DestinationNode } from '@elbwalker/source-node';
-import type { Handler, Mapping } from '@elbwalker/types';
+import type {
+  Handler,
+  Mapping,
+  Destination as WalkerOSDestination,
+} from '@elbwalker/types';
 
 export interface Destination
   extends DestinationNode.Destination<Custom, CustomEvent> {
   init: InitFn;
 }
 
-export type PushFn = DestinationNode.PushFn<Custom, CustomEvent>;
+export type PushFn = WalkerOSDestination.PushFn<Custom, CustomEvent>;
 export type InitFn = DestinationNode.InitFn<PartialConfig, Config>;
 
 export type Config = {

packages/sources/node/src/index.ts

@@ -22,6 +22,7 @@ export function sourceNode(
     version,
     ...state,
     push: (() => {}) as unknown as Elb.Fn, // Placeholder for the actual push function
+    on: {}, // Initialize empty on handlers
   };
 
   // Overwrite the push function with the instance-reference

packages/sources/node/src/lib/destination.ts

@@ -1,16 +1,5 @@
-import type { Destination, Mapping, WalkerOS } from '@elbwalker/types';
 import type { SourceNode, DestinationNode } from '../types';
-import {
-  debounce,
-  getMappingEvent,
-  getId,
-  isSameType,
-  useHooks,
-  getMappingValue,
-  isDefined,
-  isObject,
-  assign,
-} from '@elbwalker/utils';
+import { getId, isSameType } from '@elbwalker/utils';
 import { pushToDestinations } from './push';
 
 export async function addDestination(
@@ -66,80 +55,3 @@ export async function destinationInit(
 
   return true; // Destination is ready to push
 }
-
-function resolveMappingData(
-  event: WalkerOS.Event,
-  data?: Mapping.Data,
-): Destination.Data {
-  if (!data) return;
-
-  return getMappingValue(event, data);
-}
-
-export async function destinationPush(
-  instance: SourceNode.Instance,
-  destination: DestinationNode.Destination,
-  event: WalkerOS.Event,
-): Promise<boolean> {
-  const { config } = destination;
-  const { eventMapping, mappingKey } = getMappingEvent(event, config.mapping);
-
-  let data = resolveMappingData(event, config.data);
-
-  if (eventMapping) {
-    // Check if event should be processed or ignored
-    if (eventMapping.ignore) return false;
-
-    // Check to use specific event names
-    if (eventMapping.name) event.event = eventMapping.name;
-
-    // Transform event to a custom data
-    if (eventMapping.data) {
-      const dataEvent = resolveMappingData(event, eventMapping.data);
-      data =
-        isObject(data) && isObject(dataEvent) // Only merge objects
-          ? assign(data, dataEvent)
-          : dataEvent;
-    }
-  }
-
-  const options = { data, instance };
-
-  if (eventMapping?.batch && destination.pushBatch) {
-    const batched = eventMapping.batched || {
-      key: mappingKey || '',
-      events: [],
-      data: [],
-    };
-    batched.events.push(event);
-    if (isDefined(data)) batched.data.push(data);
-
-    eventMapping.batchFn =
-      eventMapping.batchFn ||
-      debounce((destination, instance) => {
-        useHooks(
-          destination.pushBatch!,
-          'DestinationPushBatch',
-          instance.hooks,
-        )(batched, destination.config, options);
-
-        // Reset the batched queues
-        // pushBatch isn't async yet, may cause trouble
-        batched.events = [];
-        batched.data = [];
-      }, eventMapping.batch);
-
-    eventMapping.batched = batched;
-    eventMapping.batchFn(destination, instance);
-  } else {
-    // It's time to go to the destination's side now
-    await useHooks(destination.push, 'DestinationPush', instance.hooks)(
-      event,
-      destination.config,
-      eventMapping,
-      options,
-    );
-  }
-
-  return true;
-}