Use resolvers context to access dgraph and postgres

src/database.ts

@@ -30,6 +30,8 @@ interface IExtensions {
   storeState: StoreStateRepo;
 }
 
+export type DatabaseClient = IDatabase<IExtensions> & IExtensions & IMain;
+
 // pg-promise initialization options:
 const initOptions: IOptions<IExtensions> = {
   promiseLib: promise,
@@ -60,7 +62,7 @@ const config = {
 const pgp: IMain = pgPromise(initOptions);
 
 // Create the database instance with extensions:
-const db = pgp(config) as IDatabase<IExtensions> & IExtensions & IMain;
+const db = pgp(config) as DatabaseClient;
 
 // Load and initialize optional diagnostics:
 import diagnostics = require("./util/db/diagnostics");

src/dgraph_import.ts

@@ -1,6 +1,6 @@
 import parseArgv from "minimist";
 import { Cursor, ICursorResult } from "./ingest/cursor";
-import { Connection } from "./storage/connection";
+import { Dgraph } from "./storage/dgraph";
 import logger from "./util/logger";
 import "./util/memo";
 import { DGRAPH_URL } from "./util/secrets";
@@ -21,7 +21,7 @@ try {
   process.exit(-1);
 }
 
-const c = new Connection();
+const c = new Dgraph();
 
 setStellarNetwork().then((network: string) => {
   logger.info(`Using ${network}`);

src/graphql.ts

@@ -4,9 +4,10 @@ import { ApolloError, ApolloServer } from "apollo-server";
 import { GraphQLError } from "graphql";
 import Honeybadger from "honeybadger";
 
+import { db } from "./database";
 import { Cursor, Worker } from "./ingest";
 import schema from "./schema";
-import { Connection } from "./storage/connection";
+import { Dgraph } from "./storage/dgraph";
 import logger from "./util/logger";
 import { BIND_ADDRESS, DEBUG_LEDGER, DGRAPH_URL, INGEST_INTERVAL, PORT } from "./util/secrets";
 import { setNetwork as setStellarNetwork } from "./util/stellar";
@@ -18,6 +19,7 @@ const server = new ApolloServer({
   playground: true,
   debug: true,
   cors: true,
+  context: () => ({ dgraph: new Dgraph(), db }),
   formatError: (error: ApolloError) => {
     logger.error(error);
 
@@ -43,7 +45,7 @@ setStellarNetwork().then((network: string) => {
 
 if (DGRAPH_URL) {
   logger.info(`[DGraph] Updating schema...`);
-  new Connection().migrate().catch((err: any) => {
+  new Dgraph().migrate().catch((err: any) => {
     logger.error(err);
     Honeybadger.notify(err);
     process.exit(-1);

src/ingest/worker.ts

@@ -3,7 +3,7 @@ import { Cursor } from "./cursor";
 
 import { SubscriptionPayloadCollection } from "./subscription_payload_collection";
 
-import { Connection } from "../storage/connection";
+import { Dgraph } from "../storage/dgraph";
 import { DGRAPH_URL } from "../util/secrets";
 
 export class Worker {
@@ -23,7 +23,7 @@ export class Worker {
       await Publisher.publish(header, collection);
 
       if (DGRAPH_URL) {
-        const c = new Connection();
+        const c = new Dgraph();
         await c.importLedgerTransactions(header, transactions);
         await c.importLedgerState(header, transactions);
         c.close();

src/schema/resolvers/account.ts

@@ -1,18 +1,17 @@
+import _ from "lodash";
 import { Account, DataEntry, Signer, TrustLine } from "../../model";
 
 import { withFilter } from "graphql-subscriptions";
 import { createBatchResolver, eventMatches, ledgerResolver } from "./util";
 
-import { db } from "../../database";
 import { joinToMap } from "../../util/array";
 
 import { ACCOUNT, pubsub } from "../../pubsub";
+import { IApolloContext } from "../../util/types";
 
-const fetchIDs = (r: any) => r.id;
-
-const signersResolver = createBatchResolver<Account, Signer[]>(async (source: any) => {
-  const accountIDs = source.map(fetchIDs);
-  const signers = await db.signers.findAllByAccountIDs(accountIDs);
+const signersResolver = createBatchResolver<Account, Signer[]>(async (source: any, args: any, ctx: IApolloContext) => {
+  const accountIDs = _.map(source, "id");
+  const signers = await ctx.db.signers.findAllByAccountIDs(accountIDs);
 
   const map = joinToMap(accountIDs, signers);
 
@@ -30,23 +29,25 @@ const signersResolver = createBatchResolver<Account, Signer[]>(async (source: an
   return signers;
 });
 
-const dataEntriesResolver = createBatchResolver<Account, DataEntry[]>((source: any) =>
-  db.dataEntries.findAllByAccountIDs(source.map(fetchIDs))
+const dataEntriesResolver = createBatchResolver<Account, DataEntry[]>((source: any, args: any, ctx: IApolloContext) =>
+  ctx.db.dataEntries.findAllByAccountIDs(_.map(source, "id"))
 );
 
-const trustLinesResolver = createBatchResolver<Account, TrustLine[]>(async (source: any) => {
-  const accountIDs = source.map(fetchIDs);
-  const trustLines = await db.trustLines.findAllByAccountIDs(accountIDs);
+const trustLinesResolver = createBatchResolver<Account, TrustLine[]>(
+  async (source: any, args: any, ctx: IApolloContext) => {
+    const accountIDs = _.map(source, "id");
+    const trustLines = await ctx.db.trustLines.findAllByAccountIDs(accountIDs);
 
-  const map = joinToMap(accountIDs, trustLines);
+    const map = joinToMap(accountIDs, trustLines);
 
-  for (const [accountID, accountTrustLines] of map) {
-    const account = source.find((acc: Account) => acc.id === accountID);
-    accountTrustLines.unshift(TrustLine.buildFakeNative(account));
-  }
+    for (const [accountID, accountTrustLines] of map) {
+      const account = source.find((acc: Account) => acc.id === accountID);
+      accountTrustLines.unshift(TrustLine.buildFakeNative(account));
+    }
 
-  return trustLines;
-});
+    return trustLines;
+  }
+);
 
 const accountSubscription = (event: string) => {
   return {
@@ -57,14 +58,14 @@ const accountSubscription = (event: string) => {
       }
     ),
 
-    resolve(payload: any, args: any, ctx: any, info: any) {
+    resolve(payload: any, args: any, ctx: IApolloContext, info: any) {
       return payload;
     }
   };
 };
 
-const signerForResolver = async (subject: Account, args: any) => {
-  const accounts = db.accounts.findAllBySigner(subject.id, args.first);
+const signerForResolver = async (subject: Account, args: any, ctx: IApolloContext) => {
+  const accounts = ctx.db.accounts.findAllBySigner(subject.id, args.first);
   return [subject].concat(await accounts);
 };
 
@@ -77,20 +78,20 @@ export default {
     signerFor: signerForResolver
   },
   Query: {
-    account(root: any, args: any, ctx: any, info: any) {
-      return db.accounts.findByID(args.id);
+    account(root: any, args: any, ctx: IApolloContext, info: any) {
+      return ctx.db.accounts.findByID(args.id);
     },
-    accounts(root: any, args: any, ctx: any, info: any) {
-      return db.accounts.findAllByIDs(args.id);
+    accounts(root: any, args: any, ctx: IApolloContext, info: any) {
+      return ctx.db.accounts.findAllByIDs(args.id);
     },
-    async accountsSignedBy(root: any, args: any, ctx: any, info: any) {
-      const account = await db.accounts.findByID(args.id);
+    async accountsSignedBy(root: any, args: any, ctx: IApolloContext, info: any) {
+      const account = await ctx.db.accounts.findByID(args.id);
 
       if (!account) {
         return [];
       }
 
-      return [account].concat(await db.accounts.findAllBySigner(args.id, args.first));
+      return [account].concat(await ctx.db.accounts.findAllBySigner(args.id, args.first));
     }
   },
   Subscription: {

src/schema/resolvers/asset.ts

@@ -1,9 +1,9 @@
-import { db } from "../../database";
+import { IApolloContext } from "../../util/types";
 
 export default {
   Query: {
-    async assets(root: any, args: any, ctx: any, info: any) {
-      const data = await db.assets.findAll(args.code, args.issuer, args.first, args.offset);
+    async assets(root: any, args: any, ctx: IApolloContext, info: any) {
+      const data = await ctx.db.assets.findAll(args.code, args.issuer, args.first, args.offset);
 
       return data.map(asset => {
         return {

src/schema/resolvers/data_entry.ts

@@ -3,12 +3,11 @@ import { Account, DataEntry } from "../../model";
 import { withFilter } from "graphql-subscriptions";
 import { createBatchResolver, eventMatches, ledgerResolver } from "./util";
 
-import { db } from "../../database";
-
 import { DATA_ENTRY, pubsub } from "../../pubsub";
+import { IApolloContext } from "../../util/types";
 
-const accountResolver = createBatchResolver<DataEntry, Account>((source: any) =>
-  db.accounts.findAllByIDs(source.map((r: DataEntry) => r.accountID))
+const accountResolver = createBatchResolver<DataEntry, Account>((source: any, args: any, ctx: IApolloContext) =>
+  ctx.db.accounts.findAllByIDs(source.map((r: DataEntry) => r.accountID))
 );
 
 const dataEntrySubscription = (event: string) => {
@@ -20,7 +19,7 @@ const dataEntrySubscription = (event: string) => {
       }
     ),
 
-    resolve(payload: any, args: any, ctx: any, info: any) {
+    resolve(payload: any, args: any, ctx: IApolloContext, info: any) {
       return payload;
     }
   };
@@ -35,8 +34,8 @@ export default {
     dataEntry: dataEntrySubscription(DATA_ENTRY)
   },
   Query: {
-    dataEntries(root: any, args: any, ctx: any, info: any) {
-      return db.dataEntries.findAllByAccountID(args.id);
+    dataEntries(root: any, args: any, ctx: IApolloContext, info: any) {
+      return ctx.db.dataEntries.findAllByAccountID(args.id);
     }
   }
 };

src/schema/resolvers/ledger.ts

@@ -1,29 +1,29 @@
-import { db } from "../../database";
 import { Ledger, LedgerHeader } from "../../model";
 import { pubsub } from "../../pubsub";
+import { IApolloContext } from "../../util/types";
 import { createBatchResolver } from "./util";
 
 const LEDGER_CREATED = "LEDGER_CREATED";
 
-const ledgerHeaderResolver = createBatchResolver<Ledger, LedgerHeader>((source: any) =>
-  db.ledgerHeaders.findAllBySeq(source.map((r: Ledger) => r.seq))
+const ledgerHeaderResolver = createBatchResolver<Ledger, LedgerHeader>((source: any, args: any, ctx: IApolloContext) =>
+  ctx.db.ledgerHeaders.findAllBySeq(source.map((r: Ledger) => r.seq))
 );
 
 export default {
   Ledger: {
     header: ledgerHeaderResolver
   },
   Query: {
-    ledger(root: any, args: any, ctx: any, info: any) {
+    ledger(root: any, args: any, ctx: IApolloContext, info: any) {
       return new Ledger(args.seq);
     },
-    ledgers(root: any, args: any, ctx: any, info: any) {
+    ledgers(root: any, args: any, ctx: IApolloContext, info: any) {
       return args.seq.map((seq: number) => new Ledger(seq));
     }
   },
   Subscription: {
     ledgerCreated: {
-      resolve(payload: any, args: any, ctx: any, info: any) {
+      resolve(payload: any, args: any, ctx: IApolloContext, info: any) {
         return payload;
       },
       subscribe() {

src/schema/resolvers/offer.ts

@@ -4,12 +4,11 @@ import Asset from "../../util/asset";
 import { withFilter } from "graphql-subscriptions";
 import { assetResolver, createBatchResolver, eventMatches, ledgerResolver } from "./util";
 
-import { db } from "../../database";
-
 import { OFFER, pubsub } from "../../pubsub";
+import { IApolloContext } from "../../util/types";
 
-const accountResolver = createBatchResolver<Offer, Account>((source: any) =>
-  db.accounts.findAllByIDs(source.map((r: Offer) => r.sellerid))
+const accountResolver = createBatchResolver<Offer, Account>((source: any, args: any, ctx: IApolloContext) =>
+  ctx.db.accounts.findAllByIDs(source.map((r: Offer) => r.sellerid))
 );
 
 const assetFromArg = (arg: IAssetInput): Asset | null => {
@@ -51,7 +50,7 @@ const offerSubscription = (event: string) => {
       (payload, variables) => offerMatches(variables, payload)
     ),
 
-    resolve(payload: any, args: any, ctx: any, info: any) {
+    resolve(payload: any, args: any, ctx: IApolloContext, info: any) {
       return payload;
     }
   };
@@ -65,8 +64,8 @@ export default {
     ledger: ledgerResolver
   },
   Query: {
-    offers(root: any, args: any, ctx: any, info: any) {
-      return db.offers.findAll(args.seller, args.selling, args.buying, args.first, args.limit);
+    offers(root: any, args: any, ctx: IApolloContext, info: any) {
+      return ctx.db.offers.findAll(args.seller, args.selling, args.buying, args.first, args.limit);
     }
   },
   Subscription: {

src/schema/resolvers/operation.ts

@@ -1,11 +1,11 @@
 import _ from "lodash";
-import { Connection } from "../../storage/connection";
 import { OperationsQuery } from "../../storage/queries/operations";
 import { OperationKinds } from "../../storage/queries/operations/types";
+import { IApolloContext } from "../../util/types";
 
 export default {
   IOperation: {
-    __resolveType(obj: any, context: any, info: any) {
+    __resolveType(obj: any, context: IApolloContext, info: any) {
       switch (obj.kind) {
         case OperationKinds.Payment:
           return "PaymentOperation";
@@ -33,11 +33,10 @@ export default {
     }
   },
   Query: {
-    operations(root: any, args: any, ctx: any, info: any) {
+    operations(root: any, args: any, ctx: IApolloContext, info: any) {
       const { account, first, offset, filters } = args;
-      const conn = new Connection();
 
-      const query = new OperationsQuery(conn, account, filters, first, offset);
+      const query = new OperationsQuery(ctx.dgraph, account, filters, first, offset);
 
       return query.call();
     }

src/schema/resolvers/signer.ts

@@ -1,13 +1,13 @@
-import { db } from "../../database";
 import { Account, Signer } from "../../model";
+import { IApolloContext } from "../../util/types";
 import { createBatchResolver } from "./util";
 
-const accountResolver = createBatchResolver<Signer, Account>((source: any) =>
-  db.accounts.findAllByIDs(source.map((r: Signer) => r.accountID))
+const accountResolver = createBatchResolver<Signer, Account>((source: any, args: any, ctx: IApolloContext) =>
+  ctx.db.accounts.findAllByIDs(source.map((r: Signer) => r.accountID))
 );
 
-const signerResolver = createBatchResolver<Signer, Account>((source: any) =>
-  db.accounts.findAllByIDs(source.map((r: Signer) => r.signer))
+const signerResolver = createBatchResolver<Signer, Account>((source: any, args: any, ctx: IApolloContext) =>
+  ctx.db.accounts.findAllByIDs(source.map((r: Signer) => r.signer))
 );
 
 export default {
@@ -16,11 +16,11 @@ export default {
     signer: signerResolver
   },
   Query: {
-    async signers(root: any, args: any, ctx: any, info: any) {
-      const account = await db.accounts.findByID(args.id);
+    async signers(root: any, args: any, ctx: IApolloContext, info: any) {
+      const account = await ctx.db.accounts.findByID(args.id);
 
       if (account !== null) {
-        const signers = await db.signers.findAllByAccountID(args.id);
+        const signers = await ctx.db.signers.findAllByAccountID(args.id);
 
         signers.unshift(
           new Signer({