fix: run webhooks (#82)

Author: birkir

packages/prime-core/src/entities/Webhook.ts

@@ -2,7 +2,6 @@ import { User } from '@accounts/typeorm';
 import GraphQLJSON from 'graphql-type-json';
 import { Field, ID, ObjectType } from 'type-graphql';
 import {
-  AfterInsert,
   Column,
   CreateDateColumn,
   Entity,
@@ -49,9 +48,4 @@ export class Webhook {
 
   @OneToMany(type => WebhookCall, call => call.webhook)
   public calls: WebhookCall[];
-
-  @AfterInsert()
-  public notify() {
-    // pubSub.publish('WEBHOOK_ADDED', this);
-  }
 }

packages/prime-core/src/entities/WebhookCall.ts

@@ -18,7 +18,7 @@ export class WebhookCall {
   @Field(type => Int)
   public status: number;
 
-  @Column('jsonb', { default: {} })
+  @Column('jsonb', { default: {}, nullable: true })
   @Field(type => GraphQLJSON)
   public request: any;
 
@@ -30,6 +30,9 @@ export class WebhookCall {
   @Field(type => Date)
   public executedAt: Date;
 
+  @Column({ nullable: true })
+  public webhookId: string;
+
   @ManyToOne(type => Webhook, webhook => webhook.calls, {
     onDelete: 'CASCADE',
   })

packages/prime-core/src/modules/internal/repositories/DocumentRepository.ts

@@ -68,6 +68,6 @@ export class DocumentRepository extends DataLoaderRepository<Document> {
     document.userId = userId;
     const res = await this.insert(document);
     const doc = res.identifiers.pop() || { id: null };
-    return this.findOneOrFail(doc.id);
+    return doc.id;
   }
 }

packages/prime-core/src/modules/internal/resolvers/DocumentResolver.ts

@@ -18,6 +18,7 @@ import { Document } from '../../../entities/Document';
 import { Context } from '../../../interfaces/Context';
 import { DocumentTransformer } from '../../../utils/DocumentTransformer';
 import { getUniqueHashId } from '../../../utils/getUniqueHashId';
+import { processWebhooks } from '../../../utils/processWebhooks';
 import { DocumentRepository } from '../repositories/DocumentRepository';
 import { SchemaFieldRepository } from '../repositories/SchemaFieldRepository';
 import { SchemaRepository } from '../repositories/SchemaRepository';
@@ -186,18 +187,18 @@ export class DocumentResolver {
     @Arg('locale', { nullable: true }) locale?: string,
     @Arg('releaseId', type => ID, { nullable: true }) releaseId?: string
   ): Promise<boolean> {
-    const doc = await this.Document(id, locale, releaseId);
+    const document = await this.Document(id, locale, releaseId);
     await this.documentRepository.update(
       {
-        documentId: doc.documentId,
+        documentId: document.documentId,
         ...(locale && { locale }),
         ...(releaseId && { releaseId }),
       },
       {
         deletedAt: new Date(),
       }
     );
-    // @todo run webhook
+    processWebhooks('document.removed', { document });
     // @todo update algolia
     return true;
   }
@@ -209,10 +210,11 @@ export class DocumentResolver {
     @Ctx() context: Context //
   ) {
     const doc = await this.documentRepository.findOneOrFail({ id, deletedAt: IsNull() });
-    const result = this.documentRepository.publish(doc, context.user.id);
-    // @todo run webhook
+    const publishedId = await this.documentRepository.publish(doc, context.user.id);
+    const document = await this.Document(publishedId);
+    processWebhooks('document.published', { document });
     // @todo update algolia
-    return result;
+    return document;
   }
 
   @Mutation(returns => Document)
@@ -230,9 +232,10 @@ export class DocumentResolver {
         publishedAt: null as any,
       }
     );
-    // @todo run webhook
     // @todo update algolia
-    return this.Document(id);
+    const document = await this.Document(id);
+    processWebhooks('document.unpublished', { document });
+    return document;
   }
 
   @FieldResolver(returns => GraphQLJSON, { nullable: true })

packages/prime-core/src/modules/internal/resolvers/ReleaseResolver.ts

@@ -4,6 +4,7 @@ import { getRepository, In } from 'typeorm';
 import { InjectRepository } from 'typeorm-typedi-extensions';
 import { Document } from '../../../entities/Document';
 import { Release } from '../../../entities/Release';
+import { processWebhooks } from '../../../utils/processWebhooks';
 import { DocumentRepository } from '../repositories/DocumentRepository';
 import { ReleaseRepository } from '../repositories/ReleaseRepository';
 import { ConnectionArgs, createConnectionType } from '../types/createConnectionType';
@@ -46,9 +47,10 @@ export class ReleaseResolver {
     @Arg('input') input: ReleaseInput,
     @Ctx() context: Context
   ): Promise<Release> {
-    const entity = this.releaseRepository.create(input);
-    await this.releaseRepository.save(entity);
-    return entity;
+    const release = this.releaseRepository.create(input);
+    await this.releaseRepository.save(release);
+    processWebhooks('release.created', { release });
+    return release;
   }
 
   @Authorized()
@@ -60,6 +62,7 @@ export class ReleaseResolver {
   ): Promise<Release> {
     const release = await this.releaseRepository.findOneOrFail(id);
     await this.releaseRepository.merge(release, input);
+    processWebhooks('release.updated', { release });
     return await this.releaseRepository.save(release);
   }
 
@@ -77,6 +80,8 @@ export class ReleaseResolver {
 
     await this.releaseRepository.remove(release);
 
+    processWebhooks('release.removed', { release });
+
     return true;
   }
 
@@ -125,6 +130,8 @@ export class ReleaseResolver {
     release.publishedBy = context.user.id;
     await this.releaseRepository.save(release);
 
+    processWebhooks('release.published', { release });
+
     return release;
   }
 

packages/prime-core/src/modules/internal/resolvers/SchemaResolver.ts

@@ -5,6 +5,7 @@ import { EntityConnection } from 'typeorm-cursor-connection';
 import { InjectRepository } from 'typeorm-typedi-extensions';
 import { Document } from '../../../entities/Document';
 import { Schema, SchemaVariant } from '../../../entities/Schema';
+import { processWebhooks } from '../../../utils/processWebhooks';
 import { pubSub } from '../../../utils/pubSub';
 import { SchemaRepository } from '../repositories/SchemaRepository';
 import { ConnectionArgs, createConnectionType } from '../types/createConnectionType';
@@ -78,6 +79,7 @@ export class SchemaResolver {
     }
     pubSub.publish('REBUILD_EXTERNAL', schema);
     schema.variant = parseEnum(SchemaVariant, schema.variant);
+    processWebhooks('schema.created', { schema });
     return schema;
   }
 
@@ -97,6 +99,7 @@ export class SchemaResolver {
     await this.schemaRepository.save(schema);
     schema.variant = parseEnum(SchemaVariant, schema.variant);
     pubSub.publish('REBUILD_EXTERNAL', schema);
+    processWebhooks('schema.updated', { schema });
     return schema;
   }
 
@@ -106,6 +109,7 @@ export class SchemaResolver {
     const schema = await this.schemaRepository.findOneOrFail(id);
     await this.schemaRepository.remove(schema);
     pubSub.publish('REBUILD_EXTERNAL', schema);
+    processWebhooks('schema.removed', { schema });
     return true;
   }
 

packages/prime-core/src/modules/internal/resolvers/UserResolver.ts

@@ -6,6 +6,7 @@ import { Arg, Args, Ctx, FieldResolver, ID, Mutation, Query, Resolver, Root } fr
 import { Repository } from 'typeorm';
 import { InjectRepository } from 'typeorm-typedi-extensions';
 import { Context } from '../../../interfaces/Context';
+import { processWebhooks } from '../../../utils/processWebhooks';
 import { UserRepository } from '../repositories/UserRepository';
 import { ConnectionArgs, createConnectionType } from '../types/createConnectionType';
 import { UpdateUserInput } from '../types/UpdateUserInput';
@@ -80,6 +81,8 @@ export class UserResolver {
       await password.server.setProfile(userId, profile);
     }
 
+    processWebhooks('user.created', { userId });
+
     return true;
   }
 
@@ -100,6 +103,7 @@ export class UserResolver {
     if (user) {
       await accounts.addEmail(user.id, email, false);
       await accounts.sendVerificationEmail(email);
+      processWebhooks('user.emailAdded', { userId: user.id, email });
       return true;
     }
     return false;
@@ -115,6 +119,7 @@ export class UserResolver {
     user.profile = input;
     context.ability.throwUnlessCan('update', user);
     await this.userRepository.save(user);
+    processWebhooks('user.updated', { user });
     return user;
   }
 
@@ -127,6 +132,7 @@ export class UserResolver {
     const user = await this.userRepository.findOneOrFail(id);
     context.ability.throwUnlessCan('delete', user);
     await this.userRepository.remove(user);
+    processWebhooks('user.removed', { user });
     return true;
   }
 

packages/prime-core/src/utils/processWebhooks.ts

@@ -0,0 +1,74 @@
+import { getRepository } from 'typeorm';
+import { Webhook } from '../entities/Webhook';
+import { WebhookCall } from '../entities/WebhookCall';
+
+const successCodes = [200, 201, 202, 203, 204];
+
+export const processWebhooks = async (action, body) => {
+  const webhookRepository = getRepository(Webhook);
+  const webhookCallRepository = getRepository(WebhookCall);
+
+  const webhooks = await webhookRepository.find();
+
+  await Promise.all(
+    webhooks
+      .filter(webhook => {
+        if (webhook.options && Array.isArray(webhook.options.actions)) {
+          return webhook.options.actions.includes(action);
+        }
+        return true;
+      })
+      .map(async webhook => {
+        const request: any = {
+          headers: {
+            'x-prime-action': action,
+            'x-prime-webhook-name': webhook.name,
+            'content-type': 'application/json',
+            'user-agent': 'prime',
+          },
+          method: webhook.method,
+          url: webhook.url,
+          body,
+        };
+
+        let response;
+
+        try {
+          response = await fetch(request.url, {
+            headers: request.headers,
+            method: request.method,
+            body: JSON.stringify(request.body),
+          });
+        } catch (err) {
+          response = {
+            headers: [],
+            status: -1,
+            statusText: err.message,
+            text: async () => '',
+          };
+        }
+
+        const headers = {};
+        response.headers.forEach((value, key) => {
+          headers[key] = value;
+        });
+
+        await webhookCallRepository.create({
+          webhookId: webhook.id,
+          success: successCodes.indexOf(response.status) >= 0,
+          status: response.status,
+          executedAt: new Date(),
+          request,
+          response: {
+            headers,
+            body: await response.text(),
+            redirected: response.redirected,
+            url: response.url,
+            type: response.type,
+            status: response.status,
+            statusText: response.statusText,
+          },
+        });
+      })
+  );
+};