Send parsed game events to events processor(untested)

src/core/game_events/betika/index.ts

@@ -0,0 +1,43 @@
+import { BaseGameEventsProcessor } from "..";
+import { getConfig } from "../../..";
+import { BetProvider } from "../../../bet_providers";
+import { BetikaProvider } from "../../../bet_providers/betika";
+import { RedisSingleton } from "../../../datastores/redis";
+import { getRedisProcessedEventsChannelName } from "../../../utils/redis";
+import { ProcessedGameEvents } from "../../../utils/types/common";
+import { Result } from "../../../utils/types/result_type";
+
+const {logger} = getConfig();
+
+export class BetikaGameEventsProcessor extends BaseGameEventsProcessor {
+    public override betProvider: BetProvider;
+
+    constructor() {
+        super();
+        this.betProvider = new BetikaProvider();
+    }
+
+    public async subscribeToChannels(): Promise<Result<boolean, Error>> {
+        const getBetProviderConfigResult = await this.betProvider.getConfig();
+        if (getBetProviderConfigResult.result === "error") {
+            logger.error("Events processor failed to load config for provider: ", this.betProvider.name);
+            return getBetProviderConfigResult;
+        }
+
+        const getRedisSubscriberResult = await RedisSingleton.getSubscriber();
+        if (getRedisSubscriberResult.result === "success") {
+            const betProviderConfig = getBetProviderConfigResult.value;
+            const results = betProviderConfig.games.map(async game => {
+                await getRedisSubscriberResult.value.subscribe(getRedisProcessedEventsChannelName(this.betProvider, game.name, game.betType), message => {
+                    const parsedMessage = JSON.parse(message) as ProcessedGameEvents;
+                    logger.trace("redis subscriber message received. ", parsedMessage)
+                });
+            });
+            await Promise.all(results);
+            return {result: "success", value: true};
+        } else {
+            logger.error("Events processor failed to connect to redis subscriber for provider: ", this.betProvider.name);
+            return getRedisSubscriberResult;
+        }
+    }
+}

src/core/game_events/index.ts

@@ -0,0 +1,5 @@
+import { BetProvider } from "../../bet_providers";
+
+export abstract class BaseGameEventsProcessor {
+    public abstract betProvider: BetProvider
+}

src/core/parsers/betika/index.ts

@@ -3,8 +3,8 @@ import { getConfig } from "../../..";
 import { BetProvider } from "../../../bet_providers";
 import { BetikaProvider } from "../../../bet_providers/betika";
 import { RedisSingleton } from "../../../datastores/redis";
-import { getRedisHtmlParserChannelName } from "../../../utils/redis";
-import { BetTypes, RawHtmlForProcessingMessage } from "../../../utils/types/common";
+import { getRedisProcessedEventsChannelName, getRedisHtmlParserChannelName } from "../../../utils/redis";
+import { BetTypes, ProcessedThreeWayGameEvent, ProcessedTwoWayGameEvent, RawHtmlForProcessingMessage } from "../../../utils/types/common";
 import { Result } from "../../../utils/types/result_type";
 import { processBetikaThreeWayGamesHtml, processBetikaTwoWayGamesHtml } from "./parser_types";
 
@@ -57,14 +57,46 @@ export class BetikaParser extends BaseParser {
         }
     }
 
-    private processRawHtmlMessage(parsedMessage: RawHtmlForProcessingMessage): void {
+    private async processRawHtmlMessage(parsedMessage: RawHtmlForProcessingMessage): Promise<void> {
         let results2;
+        let parsedResults: ProcessedTwoWayGameEvent[] | ProcessedThreeWayGameEvent[];
         switch (parsedMessage.betType) {
             case BetTypes.TWO_WAY:
                 results2 = processBetikaTwoWayGamesHtml(parsedMessage.rawHtml);
+                if (results2.result === "success") {
+                    parsedResults = results2.value.map(item => {
+                        return {
+                            oddsAWin: item.oddsAWin,
+                            oddsBWin: item.oddsBWin,
+                            league: item.league,
+                            estimatedStartTimeUtc: item.estimatedStartTimeUtc,
+                            meta: JSON.stringify({
+                                link: item.link
+                            })
+                        } as ProcessedTwoWayGameEvent;
+                    });
+                } else {
+                    throw new Error("Failed to process Betika two way games html");
+                }
                 break;
             case BetTypes.THREE_WAY:
                 results2 = processBetikaThreeWayGamesHtml(parsedMessage.rawHtml);
+                if (results2.result === "success") {
+                    parsedResults = results2.value.map(item => {
+                        return {
+                            oddsAWin: item.oddsAWin,
+                            oddsBWin: item.oddsBWin,
+                            oddsDraw: item.oddsDraw,
+                            league: item.league,
+                            estimatedStartTimeUtc: item.estimatedStartTimeUtc,
+                            meta: JSON.stringify({
+                                link: item.link
+                            })
+                        } as ProcessedThreeWayGameEvent;
+                    });
+                } else {
+                    throw new Error("Failed to process Betika Three way games html");
+                }
                 break;
             default:
                 const message = "Unknown bet type provided";
@@ -77,16 +109,28 @@ export class BetikaParser extends BaseParser {
                 throw new Error(`Unknown bet type provided for provider: ${this.betProvider.name}`);
         }
 
-        if (results2.result === "success") {
-            logger.info("Successfully fetched games: ", results2.value);
+        logger.info("Successfully fetched games: ", results2.value);
+        const getRedisPublisherResult = await RedisSingleton.getPublisher();
+
+        if (getRedisPublisherResult.result === "success") {
+            this.publishProcessedGameEvents(
+                getRedisPublisherResult.value,
+                getRedisProcessedEventsChannelName(this.betProvider, parsedMessage.gameName, parsedMessage.betType),
+                {
+                betProviderName: parsedMessage.betProviderName,
+                betType: parsedMessage.betType,
+                gameName: parsedMessage.gameName,
+                data: parsedResults
+                });
         } else {
-            logger.error("Failed to parse html into games: ", {
+            const message = "Failed to get redis publisher to send processed events: ";
+            logger.error(message, {
                 betProviderName: parsedMessage.betProviderName,
                 betType: parsedMessage.betType,
                 fromUrl: parsedMessage.fromUrl,
                 gameName: parsedMessage.gameName,
-                errorMessage: results2.value.message
-            });
+                errorMessage: getRedisPublisherResult.value.message
+            })
         }
     }
 }

src/core/parsers/index.ts

@@ -1,5 +1,15 @@
+import { RedisClientType } from "redis";
 import { BetProvider } from "../../bet_providers";
+import { ProcessedGameEvents } from "../../utils/types/common";
 
 export abstract class BaseParser {
     public abstract betProvider: BetProvider;
+
+    protected async publishProcessedGameEvents(
+        redisPublisher: RedisClientType,
+        channelName: string,
+        data: ProcessedGameEvents
+    ) {
+        await redisPublisher.publish(channelName, JSON.stringify(data));
+    }
 }

src/datastores/postgres/entities/index.ts

@@ -1,9 +1,11 @@
 import { Entity, Column, PrimaryGeneratedColumn, Index} from "typeorm";
+import { BetProviders, Games } from "../../../utils/types/common";
 
 /**
  * Table representing game events where there is no possibility of a draw.
  * A good example is tennis.
  */
+@Index(["bet_provider_id", "bet_provider_name"], {unique: true})
 @Entity({name: "two_way_game_event"})
 export class TwoWayGameEventEntity {
     @PrimaryGeneratedColumn()
@@ -25,11 +27,18 @@ export class TwoWayGameEventEntity {
     @Column("decimal", {nullable: false})
     odds_b_win: number
 
+    @Index("two_way_game_event_bet_provider_idx")
     @Column("varchar", {length: 100, nullable: false})
-    bet_provider_name: string
+    bet_provider_name: BetProviders
 
     @Column("varchar", {length: 100, nullable: false})
-    game_name: string
+    game_name: Games
+
+    @Column("varchar", {length: 100, nullable: false})
+    league: string
+
+    @Column("json", {nullable: false})
+    meta_data: string
 
     @Index("two_way_game_event_created_at_idx")
     @Column("timestamptz", {nullable: false, default: () => "CURRENT_TIMESTAMP"})
@@ -43,6 +52,7 @@ export class TwoWayGameEventEntity {
  * Table representing game events where the is possibility of a draw. A great
  * example is football.
  */
+@Index(["bet_provider_id", "bet_provider_name"], {unique: true})
 @Entity({name: "three_way_game_event"})
 export class ThreeWayGameEventEntity {
     @PrimaryGeneratedColumn()
@@ -67,11 +77,18 @@ export class ThreeWayGameEventEntity {
     @Column("decimal", {nullable: false})
     odds_draw: number
 
+    @Index("three_way_game_event_bet_provider_idx")
     @Column("varchar", {length: 100, nullable: false})
-    bet_provider: string
+    bet_provider: BetProviders
 
     @Column("varchar", {length: 100, nullable: false})
-    game_name: string
+    game_name: Games
+
+    @Column("varchar", {length: 100, nullable: false})
+    league: string
+
+    @Column("json", {nullable: false})
+    meta_data: string
 
     @Index("three_way_game_event_created_at_idx")
     @Column("timestamptz", {nullable: false, default: () => "CURRENT_TIMESTAMP"})

src/datastores/postgres/queries/three_way_game_event/index.ts

@@ -0,0 +1,65 @@
+import { DataSource, InsertResult, UpdateResult } from "typeorm";
+import { BetProviders } from "../../../../utils/types/common";
+import { ThreeWayGameEvent } from "../../../../utils/types/db";
+import { ThreeWayGameEventEntity } from "../../entities";
+
+/**
+ * Useful for checking whether a three way game event already exists for a provider.
+ * It can then either be created if not exists, updated if the odds have changed, or ignored if no changes.
+ * @param dataSource
+ * @param betProviderId 
+ * @param betProviderName 
+ * @returns 
+ */
+export const getThreeWayGame = async (
+    dataSource: DataSource,
+    betProviderId: string,
+    betProviderName: BetProviders
+): Promise<ThreeWayGameEventEntity | null> => {
+    return await dataSource.createQueryBuilder()
+    .select("three_way_game_event")
+    .from(ThreeWayGameEventEntity, "three_way_game_event")
+    .where("bet_provider_id = :betProviderId", {betProviderId: betProviderId})
+    .andWhere("bet_provider_name = :betProviderName", {betProviderName})
+    .getOne();
+};
+
+export const insertThreeWayGameEvent = async (
+    dataSource: DataSource,
+    data: ThreeWayGameEvent
+): Promise<InsertResult> => {
+    const toDataBase = {
+        bet_provider_id: data.betProviderId,
+        bet_provider_name: data.betProviderName,
+        club_a: data.clubA,
+        club_b: data.clubB,
+        odds_a_win: data.oddsAWin,
+        odds_b_win: data.oddsBWin,
+        game_name: data.gameName
+    }
+    return await dataSource.createQueryBuilder()
+    .insert()
+    .into(ThreeWayGameEventEntity)
+    .values(toDataBase)
+    .execute();
+};
+
+export const updateThreeWayGameEventOdds = async (
+    dataSource: DataSource,
+    betProviderId: string,
+    betProviderName: BetProviders,
+    oddsAWin: number,
+    oddsBWin: number,
+    oddsDraw: number
+): Promise<UpdateResult> => {
+    return await dataSource.createQueryBuilder()
+    .update(ThreeWayGameEventEntity)
+    .set({
+        odds_a_win: oddsAWin,
+        odds_b_win: oddsBWin,
+        odds_draw: oddsDraw
+    })
+    .where("bet_provider_id = :betProviderId", {betProviderId: betProviderId})
+    .andWhere("bet_provider_name = :betProviderName", {betProviderName})
+    .execute();
+};

src/datastores/postgres/queries/two_way_game_event/index.ts

@@ -0,0 +1,63 @@
+import { DataSource, InsertResult, UpdateResult } from "typeorm";
+import { BetProviders } from "../../../../utils/types/common";
+import { TwoWayGameEvent } from "../../../../utils/types/db";
+import { TwoWayGameEventEntity } from "../../entities";
+
+/**
+ * Useful for checking whether a two way game event already exists for a provider.
+ * It can then either be created if not exists, updated if the odds have changed, or ignored if no changes.
+ * @param dataSource
+ * @param betProviderId 
+ * @param betProviderName 
+ * @returns 
+ */
+export const getTwoWayGame = async (
+    dataSource: DataSource,
+    betProviderId: string,
+    betProviderName: BetProviders
+): Promise<TwoWayGameEventEntity | null> => {
+    return await dataSource.createQueryBuilder()
+    .select("two_way_game_event")
+    .from(TwoWayGameEventEntity, "two_way_game_event")
+    .where("bet_provider_id = :betProviderId", {betProviderId: betProviderId})
+    .andWhere("bet_provider_name = :betProviderName", {betProviderName})
+    .getOne();
+};
+
+export const insertTwoWayGameEvent = async (
+    dataSource: DataSource,
+    data: TwoWayGameEvent
+): Promise<InsertResult> => {
+    const toDataBase = {
+        bet_provider_id: data.betProviderId,
+        bet_provider_name: data.betProviderName,
+        club_a: data.clubA,
+        club_b: data.clubB,
+        odds_a_win: data.oddsAWin,
+        odds_b_win: data.oddsBWin,
+        game_name: data.gameName
+    };
+    return await dataSource.createQueryBuilder()
+    .insert()
+    .into(TwoWayGameEventEntity)
+    .values(toDataBase)
+    .execute();
+};
+
+export const updateTwoWayGameEventOdds = async (
+    dataSource: DataSource,
+    betProviderId: string,
+    betProviderName: BetProviders,
+    oddsAWin: number,
+    oddsBWin: number
+): Promise<UpdateResult> => {
+    return await dataSource.createQueryBuilder()
+    .update(TwoWayGameEventEntity)
+    .set({
+        odds_a_win: oddsAWin,
+        odds_b_win: oddsBWin
+    })
+    .where("bet_provider_id = :betProviderId", {betProviderId: betProviderId})
+    .andWhere("bet_provider_name = :betProviderName", {betProviderName})
+    .execute();
+};