Update sync code for unified-accounting

connectors/connector-plaid/def.ts

@@ -100,7 +100,13 @@ export const plaidSchemas = {
     })
     .default({}),
   sourceOutputEntities: R.mapToObj(
-    ['transaction', 'account', 'investment_transaction', 'holding'] as const,
+    [
+      'transaction',
+      'account',
+      'investment_transaction',
+      'holding',
+      'merchant',
+    ] as const,
     (e) => [e, z.unknown()],
   ),
   sourceOutputEntity: z.discriminatedUnion('entityName', [
@@ -114,6 +120,11 @@ export const plaidSchemas = {
       entityName: z.literal('transaction'),
       entity: zCast<plaid.Transaction | plaid.InvestmentTransaction>(),
     }),
+    z.object({
+      id: z.string(),
+      entityName: z.literal('merchant'),
+      entity: zCast<{merchant_entity_id: string; name: string}>(),
+    }),
   ]),
   webhookInput: zWebhookInput,
 

connectors/connector-plaid/server.ts

@@ -318,10 +318,29 @@ export const plaidServerConnector = {
           }
 
           cursor = res.next_cursor
+          const filteredTxns = [...res.added, ...res.modified].filter(
+            (t) => !accountIds || accountIds.includes(t.account_id),
+          )
+          const merchantNameById = Object.fromEntries(
+            filteredTxns
+              .map((t) =>
+                'merchant_entity_id' in t && t.merchant_entity_id
+                  ? ([
+                      t.merchant_entity_id as string,
+                      t.merchant_name!,
+                    ] as const)
+                  : null,
+              )
+              .filter((i) => !!i),
+          )
+
           yield [
-            ...[...res.added, ...res.modified]
-              .filter((t) => !accountIds || accountIds.includes(t.account_id))
-              .map((t) => def._opData('transaction', t.transaction_id, t)),
+            ...filteredTxns.map((t) =>
+              def._opData('transaction', t.transaction_id, t),
+            ),
+            ...Object.entries(merchantNameById).map(([id, name]) =>
+              def._opData('merchant', id, {merchant_entity_id: id, name}),
+            ),
             ...R.pipe(
               res.removed,
               R.map((t) => t.transaction_id),

packages/db/lib/upsert-sql.spec.ts

@@ -217,6 +217,7 @@ test('upsert query with inferred table', async () => {
     data: {hello: 'world'},
     arr: ['a', 'b'],
     date,
+    null_value: null,
   }
   const table = inferTableForUpsert('test_user', row)
   const createTableSql = await generateMigration(
@@ -230,7 +231,8 @@ test('upsert query with inferred table', async () => {
     	"count" text,
     	"data" jsonb,
     	"arr" jsonb,
-    	"date" text
+    	"date" text,
+    	"null_value" text
     );
     "
   `)
@@ -243,27 +245,38 @@ test('upsert query with inferred table', async () => {
     '{"hello":"world"}',
     '["a","b"]',
     date,
+    null,
   ])
   expect(await formatSql(query.toSQL().sql)).toMatchInlineSnapshot(`
     "insert into
-      "test_user" ("id", "name", "count", "data", "arr", "date")
+      "test_user" (
+        "id",
+        "name",
+        "count",
+        "data",
+        "arr",
+        "date",
+        "null_value"
+      )
     values
-      ($1, $2, $3, $4, $5, $6)
+      ($1, $2, $3, $4, $5, $6, $7)
     on conflict ("id") do
     update
     set
       "name" = excluded."name",
       "count" = excluded."count",
       "data" = excluded."data",
       "arr" = excluded."arr",
-      "date" = excluded."date"
+      "date" = excluded."date",
+      "null_value" = excluded."null_value"
     where
       (
         "test_user"."name" IS DISTINCT FROM excluded."name"
         or "test_user"."count" IS DISTINCT FROM excluded."count"
         or "test_user"."data" IS DISTINCT FROM excluded."data"
         or "test_user"."arr" IS DISTINCT FROM excluded."arr"
         or "test_user"."date" IS DISTINCT FROM excluded."date"
+        or "test_user"."null_value" IS DISTINCT FROM excluded."null_value"
       )
     "
   `)

unified/unified-accounting/adapters/plaid-adapter/mapper.ts

@@ -41,6 +41,7 @@ export const mappers = {
   transaction: mapper(zCast<Plaid['Transaction']>(), unified.transaction, {
     id: 'transaction_id',
     date: 'date',
+    vendor_id: 'merchant_entity_id',
     account_id: 'account_id',
     amount: 'amount',
     currency: 'iso_currency_code',
@@ -55,8 +56,23 @@ export const mappers = {
     // Plaid transactions are single entry and therefore unbalanced
     // We should probably not set any "lines" at all in this case.
     // Would also need some semantic to communicate set if null, aka set default value,
-    lines: () => [],
+    lines: (t) => [
+      {
+        account_id: '',
+        amount: t.amount * -1,
+        currency: t.iso_currency_code ?? t.unofficial_currency_code ?? 'USD',
+      },
+    ],
     // created_at: (t) => t.datetime ?? new Date().toISOString(),
     // updated_at: (t) => t.datetime ?? new Date().toISOString(),
   }),
+  merchant: mapper(
+    zCast<{merchant_entity_id: string; name: string}>(),
+    unified.vendor,
+    {
+      id: 'merchant_entity_id',
+      name: 'name',
+      url: () => null,
+    },
+  ),
 }

unified/unified-accounting/adapters/qbo-adapter/mapper.ts

@@ -1,6 +1,6 @@
 import type {QBO_ENTITY_NAME} from '@openint/connector-qbo'
 import {type QBOSDKTypes} from '@openint/connector-qbo'
-import {snakeCase, type z} from '@openint/util'
+import {R, snakeCase, type z} from '@openint/util'
 import {mapper, zCast} from '@openint/vdk'
 import * as unified from '../../unifiedModels'
 
@@ -36,6 +36,10 @@ const tranactionMappers = {
       }))
     },
     bank_category: () => 'Purchase',
+    vendor_id: (t) =>
+      t.EntityRef?.type === 'Vendor' && t.EntityRef?.value
+        ? makeQboId('Vendor', t.EntityRef.value)
+        : '', // for some reason `null` causes vendor_id to be quoted, probably encoded as json, does not add up but...
     created_at: 'MetaData.CreateTime',
     updated_at: 'MetaData.LastUpdatedTime',
   }),
@@ -46,17 +50,33 @@ const tranactionMappers = {
     currency: 'CurrencyRef.value',
     memo: 'PrivateNote',
     amount: 'TotalAmt',
-    lines: (p) =>
-      p.Line.map((line) => ({
-        id: line.Id,
-        memo: line.Description,
-        amount: -1 * line.Amount,
-        currency: p.CurrencyRef.value,
-        account_id: makeQboId(
-          'Account',
-          line.DepositLineDetail?.AccountRef?.value ?? '',
-        ),
-      })),
+    lines: (_t) => {
+      const t = _t as typeof _t & {
+        CashBack?: {
+          Amount: number
+          Memo: string
+          AccountRef: {value: string; name: string}
+        }
+      }
+      return R.compact([
+        ...t.Line.map((line) => ({
+          id: line.Id,
+          memo: line.Description,
+          amount: -1 * line.Amount,
+          currency: t.CurrencyRef.value,
+          account_id: makeQboId(
+            'Account',
+            line.DepositLineDetail?.AccountRef?.value ?? '',
+          ),
+        })),
+        t.CashBack && {
+          amount: t.CashBack.Amount,
+          currency: t.CurrencyRef.value,
+          memo: t.CashBack.Memo,
+          account_id: makeQboId('Account', t.CashBack.AccountRef.value),
+        },
+      ])
+    },
     bank_category: () => 'Deposit',
     created_at: 'MetaData.CreateTime',
     updated_at: 'MetaData.LastUpdatedTime',

unified/unified-accounting/unifiedAccountingLink.ts

@@ -46,6 +46,7 @@ export function unifiedAccountingLink(ctx: {
     // TODO: Should build this into the mapper itself
     const stream =
       {
+        // QBO
         Purchase: 'transaction',
         JournalEntry: 'transaction',
         Deposit: 'transaction',
@@ -55,25 +56,40 @@ export function unifiedAccountingLink(ctx: {
         Vendor: 'vendor',
         Customer: 'customer',
         Attachable: 'attachment',
+        // Plaid
+        merchant: 'vendor',
       }[op.data.entityName] ?? op.data.entityName
 
     const mapped = applyMapper(mapper, op.data.entity, {
       remote_data_key: 'remote_data',
     })
+    if ('lines' in mapped) {
+      const txn = mapped as {
+        amount?: number
+        lines: {amount: number}[]
+        id: string
+      }
+      const trialBalance = txn.lines?.reduce((acc, line) => {
+        return acc + line.amount
+      }, txn.amount ?? 0)
+      if (trialBalance !== 0) {
+        console.warn(`Transaction  does not balance: ${trialBalance} ${txn.id}`)
+      }
+    }
 
     return rxjs.of({
       ...op,
       data: {
         stream,
         data: {
           ...mapped,
-          connection_id: ctx.source.id, // Should this be the default somehow?
+          _openint_connection_id: ctx.source.id, // Should this be the default somehow?
           // Denormalize customer_id onto entities for now, though may be better
           // to just sync connection also?
-          customer_id: ctx.source.customerId,
+          _openint_customer_id: ctx.source.customerId,
         },
         upsert: {
-          key_columns: ['connection_id', 'id'],
+          key_columns: ['_openint_connection_id', 'id'],
           insert_only_columns: ['created_at'],
           no_diff_columns: ['updated_at'],
         },

unified/unified-accounting/unifiedModels.ts

@@ -17,7 +17,7 @@ const transaction_line = z.object({
   memo: z.string().nullable().optional().describe('Private line note'),
   amount: z.number().describe('(positive) debit or (negative) credit'),
   currency: currency_code,
-  account_id: z.string(),
+  account_id: z.string().describe('Empty means uncategorized'),
 })
 
 // MARK: - Models
@@ -37,6 +37,7 @@ export const transaction = z
       description:
         'Posted date for accounting purpose, may not be the same as transaction date',
     }),
+    vendor_id: z.string().nullish(),
     account_id: z.string().nullish(),
     amount: z.number().nullish(),
     currency: currency_code.nullish(),
@@ -78,7 +79,7 @@ export const vendor = z
   .object({
     ...commonFields,
     name: z.string(),
-    url: z.string(),
+    url: z.string().nullish(),
   })
   .openapi({ref: 'accounting.vendor'})