sdk changes

apps/sdk-api/src/controllers/event.controller.ts

@@ -1,4 +1,5 @@
 import { parseIp } from '@/utils/parseIp';
+import { parseReferrer } from '@/utils/parseReferrer';
 import { parseUserAgent } from '@/utils/parseUserAgent';
 import type { FastifyReply, FastifyRequest } from 'fastify';
 import { omit } from 'ramda';
@@ -56,33 +57,30 @@ export async function postEvent(
   let profileId: string | null = null;
   const projectId = request.projectId;
   const body = request.body;
-  const { path, hash, query } = parsePath(
-    body.properties?.path as string | undefined
-  );
-  const referrer = body.properties?.referrer as string | undefined;
+  const { path, hash, query } = parsePath(body.properties?.path);
+  const referrer = parseReferrer(body.properties?.referrer);
   const ip = getClientIp(request)!;
   const origin = request.headers.origin!;
   const ua = request.headers['user-agent']!;
   const uaInfo = parseUserAgent(ua);
   const salts = await getSalts();
+  const currentProfileId = generateProfileId({
+    salt: salts.current,
+    origin,
+    ip,
+    ua,
+  });
+  const previousProfileId = generateProfileId({
+    salt: salts.previous,
+    origin,
+    ip,
+    ua,
+  });
 
-  const [currentProfileId, previousProfileId, geo, eventsJobs] =
-    await Promise.all([
-      generateProfileId({
-        salt: salts.current,
-        origin,
-        ip,
-        ua,
-      }),
-      generateProfileId({
-        salt: salts.previous,
-        origin,
-        ip,
-        ua,
-      }),
-      parseIp(ip),
-      eventsQueue.getJobs(['delayed']),
-    ]);
+  const [geo, eventsJobs] = await Promise.all([
+    parseIp(ip),
+    eventsQueue.getJobs(['delayed']),
+  ]);
 
   // find session_end job
   const sessionEndJobCurrentProfileId = findJobByPrefix(
@@ -148,8 +146,9 @@ export async function postEvent(
     model: uaInfo.model,
     duration: 0,
     path: path,
-    referrer,
-    referrerName: referrer, // TODO
+    referrer: referrer.url,
+    referrerName: referrer.name,
+    referrerType: referrer.type,
   };
 
   const job = findJobByPrefix(eventsJobs, `event:${projectId}:${profileId}:`);
@@ -171,7 +170,7 @@ export async function postEvent(
           duration,
         },
       });
-      job.promote();
+      await job.promote();
     }
   }
 

apps/sdk-api/src/controllers/profile.controller.ts

@@ -0,0 +1,291 @@
+import { parseIp } from '@/utils/parseIp';
+import { parseUserAgent } from '@/utils/parseUserAgent';
+import type { FastifyReply, FastifyRequest } from 'fastify';
+import { assocPath, mergeDeepRight, path } from 'ramda';
+import { getClientIp } from 'request-ip';
+
+import { generateProfileId, toDots } from '@mixan/common';
+import type { IDBProfile, Profile } from '@mixan/db';
+import { db, getSalts } from '@mixan/db';
+import type {
+  IncrementProfilePayload,
+  UpdateProfilePayload,
+} from '@mixan/types';
+
+async function findProfile({
+  profileId,
+  ip,
+  origin,
+  ua,
+}: {
+  profileId: string | null;
+  ip: string;
+  origin: string;
+  ua: string;
+}) {
+  const salts = await getSalts();
+  const currentProfileId = generateProfileId({
+    salt: salts.current,
+    origin,
+    ip,
+    ua,
+  });
+  const previousProfileId = generateProfileId({
+    salt: salts.previous,
+    origin,
+    ip,
+    ua,
+  });
+
+  const ids = [currentProfileId, previousProfileId];
+  if (profileId) {
+    ids.push(profileId);
+  }
+
+  const profiles = await db.profile.findMany({
+    where: {
+      id: {
+        in: ids,
+      },
+    },
+  });
+
+  return profiles.find((p) => {
+    return (
+      p.id === profileId ||
+      p.id === currentProfileId ||
+      p.id === previousProfileId
+    );
+  }) as IDBProfile | undefined;
+}
+
+export async function updateProfile(
+  request: FastifyRequest<{
+    Body: UpdateProfilePayload;
+  }>,
+  reply: FastifyReply
+) {
+  const body = request.body;
+  const profileId: string | null = body.profileId ?? null;
+  const projectId = request.projectId;
+  const ip = getClientIp(request)!;
+  const origin = request.headers.origin ?? projectId;
+  const ua = request.headers['user-agent']!;
+  const salts = await getSalts();
+  const uaInfo = parseUserAgent(ua);
+  const geo = await parseIp(ip);
+
+  if (profileId === null) {
+    const currentProfileId = generateProfileId({
+      salt: salts.current,
+      origin,
+      ip,
+      ua,
+    });
+    const previousProfileId = generateProfileId({
+      salt: salts.previous,
+      origin,
+      ip,
+      ua,
+    });
+
+    const profiles = await db.profile.findMany({
+      where: {
+        id: {
+          in: [currentProfileId, previousProfileId],
+        },
+      },
+    });
+
+    if (profiles.length === 0) {
+      const profile = await db.profile.create({
+        data: {
+          id: currentProfileId,
+          external_id: body.id,
+          first_name: body.first_name,
+          last_name: body.last_name,
+          email: body.email,
+          avatar: body.avatar,
+          project_id: projectId,
+          properties: body.properties ?? {},
+          // ...uaInfo,
+          // ...geo,
+        },
+      });
+
+      return reply.status(201).send(profile);
+    }
+    const currentProfile = profiles.find((p) => p.id === currentProfileId);
+    const previousProfile = profiles.find((p) => p.id === previousProfileId);
+    const profile = currentProfile ?? previousProfile;
+
+    if (profile) {
+      await db.profile.update({
+        where: {
+          id: profile.id,
+        },
+        data: {
+          external_id: body.id,
+          first_name: body.first_name,
+          last_name: body.last_name,
+          email: body.email,
+          avatar: body.avatar,
+          properties: toDots(
+            mergeDeepRight(
+              profile.properties as Record<string, unknown>,
+              body.properties ?? {}
+            )
+          ),
+          // ...uaInfo,
+          // ...geo,
+        },
+      });
+
+      return reply.status(200).send(profile.id);
+    }
+
+    return reply.status(200).send();
+  }
+
+  const profile = await db.profile.findUnique({
+    where: {
+      id: profileId,
+    },
+  });
+
+  if (profile) {
+    await db.profile.update({
+      where: {
+        id: profile.id,
+      },
+      data: {
+        external_id: body.id,
+        first_name: body.first_name,
+        last_name: body.last_name,
+        email: body.email,
+        avatar: body.avatar,
+        properties: toDots(
+          mergeDeepRight(
+            profile.properties as Record<string, unknown>,
+            body.properties ?? {}
+          )
+        ),
+        // ...uaInfo,
+        // ...geo,
+      },
+    });
+  } else {
+    await db.profile.create({
+      data: {
+        id: profileId,
+        external_id: body.id,
+        first_name: body.first_name,
+        last_name: body.last_name,
+        email: body.email,
+        avatar: body.avatar,
+        project_id: projectId,
+        properties: body.properties ?? {},
+        // ...uaInfo,
+        // ...geo,
+      },
+    });
+  }
+
+  reply.status(202).send(profileId);
+}
+
+export async function incrementProfileProperty(
+  request: FastifyRequest<{
+    Body: IncrementProfilePayload;
+  }>,
+  reply: FastifyReply
+) {
+  const body = request.body;
+  const profileId: string | null = body.profileId ?? null;
+  const projectId = request.projectId;
+  const ip = getClientIp(request)!;
+  const origin = request.headers.origin ?? projectId;
+  const ua = request.headers['user-agent']!;
+
+  const profile = await findProfile({
+    ip,
+    origin,
+    ua,
+    profileId,
+  });
+
+  if (!profile) {
+    return reply.status(404).send('Not found');
+  }
+
+  const property = path(body.property.split('.'), profile.properties);
+
+  if (typeof property !== 'number' && typeof property !== 'undefined') {
+    return reply.status(400).send('Not number');
+  }
+
+  profile.properties = assocPath(
+    body.property.split('.'),
+    property ? property + body.value : body.value,
+    profile.properties
+  );
+
+  await db.profile.update({
+    where: {
+      id: profile.id,
+    },
+    data: {
+      properties: profile.properties as any,
+    },
+  });
+
+  reply.status(202).send(profile.id);
+}
+
+export async function decrementProfileProperty(
+  request: FastifyRequest<{
+    Body: IncrementProfilePayload;
+  }>,
+  reply: FastifyReply
+) {
+  const body = request.body;
+  const profileId: string | null = body.profileId ?? null;
+  const projectId = request.projectId;
+  const ip = getClientIp(request)!;
+  const origin = request.headers.origin ?? projectId;
+  const ua = request.headers['user-agent']!;
+
+  const profile = await findProfile({
+    ip,
+    origin,
+    ua,
+    profileId,
+  });
+
+  if (!profile) {
+    return reply.status(404).send('Not found');
+  }
+
+  const property = path(body.property.split('.'), profile.properties);
+
+  if (typeof property !== 'number') {
+    return reply.status(400).send('Not number');
+  }
+
+  profile.properties = assocPath(
+    body.property.split('.'),
+    property ? property - body.value : -body.value,
+    profile.properties
+  );
+
+  await db.profile.update({
+    where: {
+      id: profile.id,
+    },
+    data: {
+      properties: profile.properties as any,
+    },
+  });
+
+  reply.status(202).send(profile.id);
+}

apps/sdk-api/src/index.ts

@@ -7,6 +7,7 @@ import { redisPub } from '@mixan/redis';
 
 import eventRouter from './routes/event.router';
 import liveRouter from './routes/live.router';
+import profileRouter from './routes/profile.router';
 
 declare module 'fastify' {
   interface FastifyRequest {
@@ -29,6 +30,7 @@ const startServer = async () => {
     fastify.register(FastifySSEPlugin);
     fastify.decorateRequest('projectId', '');
     fastify.register(eventRouter, { prefix: '/event' });
+    fastify.register(profileRouter, { prefix: '/profile' });
     fastify.register(liveRouter, { prefix: '/live' });
     fastify.setErrorHandler((error, request, reply) => {
       fastify.log.error(error);

apps/sdk-api/src/referrers/referrers.readme.md

@@ -0,0 +1,5 @@
+# Snowplow Referer Parser
+
+The file index.ts in this dir is generated from snowplows referer database [Snowplow Referer Parser](https://github.com/snowplow-referer-parser/referer-parser).
+
+The orginal [referers.yml](https://github.com/snowplow-referer-parser/referer-parser/blob/master/resources/referers.yml) is based on Piwik's SearchEngines.php and Socials.php, copyright 2012 Matthieu Aubry and available under the GNU General Public License v3.

apps/sdk-api/src/routes/profile.router.ts

@@ -0,0 +1,37 @@
+import * as controller from '@/controllers/profile.controller';
+import { validateSdkRequest } from '@/utils/auth';
+import type { FastifyPluginCallback } from 'fastify';
+
+const eventRouter: FastifyPluginCallback = (fastify, opts, done) => {
+  fastify.addHook('preHandler', (req, reply, done) => {
+    validateSdkRequest(req.headers)
+      .then((projectId) => {
+        req.projectId = projectId;
+        done();
+      })
+      .catch((e) => {
+        reply.status(401).send();
+      });
+  });
+
+  fastify.route({
+    method: 'POST',
+    url: '/',
+    handler: controller.updateProfile,
+  });
+
+  fastify.route({
+    method: 'POST',
+    url: '/increment',
+    handler: controller.incrementProfileProperty,
+  });
+
+  fastify.route({
+    method: 'POST',
+    url: '/decrement',
+    handler: controller.decrementProfileProperty,
+  });
+  done();
+};
+
+export default eventRouter;

apps/sdk-api/src/utils/parseIp.ts

@@ -1,12 +1,34 @@
-export async function parseIp(ip: string) {
+interface RemoteIpLookupResponse {
+  country: string | undefined;
+  city: string | undefined;
+  stateprov: string | undefined;
+  continent: string | undefined;
+}
+
+interface GeoLocation {
+  country: string | undefined;
+  city: string | undefined;
+  region: string | undefined;
+  continent: string | undefined;
+}
+
+const geo: GeoLocation = {
+  country: undefined,
+  city: undefined,
+  region: undefined,
+  continent: undefined,
+};
+
+const ignore = ['127.0.0.1', '::1'];
+
+export async function parseIp(ip?: string): Promise<GeoLocation> {
+  if (!ip || ignore.includes(ip)) {
+    return geo;
+  }
+
   try {
     const geo = await fetch(`${process.env.GEO_IP_HOST}/${ip}`);
-    const res = (await geo.json()) as {
-      country: string | undefined;
-      city: string | undefined;
-      stateprov: string | undefined;
-      continent: string | undefined;
-    };
+    const res = (await geo.json()) as RemoteIpLookupResponse;
 
     return {
       country: res.country,
@@ -16,12 +38,6 @@ export async function parseIp(ip: string) {
     };
   } catch (e) {
     console.log('Failed to parse ip', e);
-
-    return {
-      country: undefined,
-      city: undefined,
-      region: undefined,
-      continent: undefined,
-    };
+    return geo;
   }
 }

apps/sdk-api/src/utils/parseReferrer.ts

@@ -0,0 +1,15 @@
+import referrers from '../referrers';
+
+export function parseReferrer(url?: string) {
+  const { hostname } = new URL(url || '');
+  const match = referrers[hostname];
+
+  console.log('Parsing referrer', url);
+  console.log('Match', match);
+
+  return {
+    name: match?.name ?? '',
+    type: match?.type ?? 'unknown',
+    url: url ?? '',
+  };
+}