[A] channel.service, channel.comp: load and send messages

src/app/api/supabase/message.service.spec.ts

@@ -0,0 +1,16 @@
+import { TestBed } from '@angular/core/testing';
+
+import { MessageService } from './message.service';
+
+describe('MessageService', () => {
+  let service: MessageService;
+
+  beforeEach(() => {
+    TestBed.configureTestingModule({});
+    service = TestBed.inject(MessageService);
+  });
+
+  it('should be created', () => {
+    expect(service).toBeTruthy();
+  });
+});

src/app/api/supabase/message.service.ts

@@ -0,0 +1,181 @@
+import { Injectable } from '@angular/core';
+import { BehaviorSubject, Observable, of, Subject } from 'rxjs';
+import { map } from 'rxjs/operators';
+import { Message } from './message';
+import { SupaService } from './supa.service';
+
+@Injectable({
+  providedIn: 'root'
+})
+export class MessageService {
+  // messageMap: Map<number, Message> = new Map();
+  messageMap = {};
+  messages: BehaviorSubject<any> = new BehaviorSubject([]);
+  subscribedChannelIds: number[] = [];
+  isListening: boolean = false;
+
+  constructor(
+    private supa: SupaService,
+  ) { }
+
+  /**
+   * Get Messages from local store.
+   * Requests data if store is emtpy.
+   * @param channel_id 
+   * @returns Observable<Message[]>
+   */
+  getMessages(channel_id: number): Observable<Message[]> {
+    if (!this.subscribedChannelIds.includes(channel_id)) {
+      this.subscribeToMessages(channel_id);
+      console.log('getMessages - REFRESH', channel_id)
+      this.supa.client.from<Message>('message').select()
+      .filter(<never>'channel_id', 'eq', channel_id)      
+      .then(data => {
+        this.updateStore(data.body);
+      })
+      .catch(error => {
+        console.error(error);
+      });
+    } else {      
+      console.log('getMessages - LOCAL', channel_id)
+    }
+    return this.messages.asObservable().pipe(
+      map(messages => messages.filter(e => e.channel_id === channel_id))
+    );
+  }
+
+  /**
+   * Listen to realtime events from message db.
+   */
+  subscribeToMessages(channel_id: number) {
+    // TODO check for this.subscribedChannelIds.includes(channel_id)
+    if (!this.isListening) {
+      this.subscribedChannelIds.push(channel_id);
+      this.supa.client.from<Message>('message').on('*', payload => {
+        console.log('subscribeToMessages - REALTIME EVENT', payload)
+        if ((payload.eventType === 'INSERT') || (payload.eventType === 'UPDATE')) {
+          this.messageMap[payload.new.id] = payload.new;
+        } else {
+          delete this.messageMap[payload.old.id];
+        }
+        this.next();
+      }).subscribe();
+    }
+  }
+
+  /** 
+   * Requests up to date data from API.
+   * Returns the local copy.
+   * @returns Observable<Message>
+   */
+  // refreshMessages(): Observable<Message[]> {
+  //   this.subscribeToMessages();
+  //   this.supa.client.from<Message>('message').select()
+  //     .then(messages => this.updateStore(messages.body))
+  //     .catch(error => console.log('Error: ', error));
+  //   return this.messages.asObservable();
+  // }
+
+  /**
+   * Update the local store with provided messages
+   * @param messages 
+   */
+  updateStore(messages: Message[]) {
+    messages.forEach(e => {
+      // this.messageMap.set(e.id, e);
+      this.messageMap[e.id] = e;
+    });
+    console.log('update messages', this.messageMap)
+    this.next();
+  }
+
+  /**
+   * Emits local store.
+   */
+  next = () => {
+    console.log('next', Object.values(this.messageMap))
+    this.messages.next(Object.values(this.messageMap))
+  };
+
+  /**
+   * Retrieve message from local store.
+   * @param id 
+   */
+  getOne(id: number) {
+    if (this.messageMap[id]) {
+      return of(this.messageMap[id]);
+    } else {
+      const subject: Subject<Message> = new Subject();
+      this.supa.client.from<Message>('message').select('id, name, description')
+      .filter(<never>'id', 'eq', id)      
+      .then(data => {
+        this.updateStore([data.body[0]]);
+        subject.next(data.body[0]);
+        subject.complete();
+      })
+      .catch(error => {
+        subject.error(error);
+        subject.complete();
+      });
+      return subject.asObservable();
+    }
+  }
+
+  /**
+   * 
+   * @param message 
+   */
+  updateOne(message: Message): Observable<Message> {
+    const subject: Subject<Message> = new Subject();
+    this.supa.client.from<Message>('message').update(message)
+      .match({ id: message.id })
+      .then(data => {
+        subject.next(data.body[0]);
+        subject.complete();
+      })
+      .catch(error => {
+        subject.error(error);
+        subject.complete();
+      });
+    return subject.asObservable();
+  }
+
+
+  /**
+   * Removes one message from db.
+   * @param message 
+   */
+  deleteOne(message: Message): Observable<Message> {
+    const subject: Subject<Message> = new Subject();
+    this.supa.client.from<Message>('message').delete()
+      .match({ id: message.id })
+      .then(data => {
+        subject.next(message);
+        subject.complete();
+      })
+      .catch(error => {
+        subject.error(error);
+        subject.complete();
+      });
+    return subject.asObservable();
+  }
+
+  /**
+   * Creates a message on the db.
+   * @param message 
+   */
+  addOne(message: Message): Observable<Message> {
+    const subject: Subject<Message> = new Subject();
+    this.supa.client.from<Message>('message').insert(message)
+      .then(data => {
+        subject.next(data.body[0]);
+        subject.complete();
+      })
+      .catch(error => {
+        subject.error(error);
+        subject.complete();
+      });
+    return subject.asObservable();
+  }
+
+}

src/app/api/supabase/message.ts

@@ -1,7 +1,10 @@
+import { User } from './user';
+
 export class Message {
     id: number;
     channel_id: number;
     user_id: string;
+    user?: User;
     message: string;
     inserted_at: string | Date;
 

src/app/app.module.ts

@@ -5,6 +5,7 @@ import { NgbModalModule } from "@ng-bootstrap/ng-bootstrap";
 import { AppRoutingModule } from './app-routing.module';
 import { FormsModule } from '@angular/forms';
 
+import { MessageService } from './api/supabase/message.service';
 import { SupaService } from './api/supabase/supa.service';
 import { StandupService } from './api/supabase/standup.service';
 import { StoryService } from './api/supabase/story.service';
@@ -42,6 +43,7 @@ import { InputManagerComponent } from './input-manager/input-manager.component';
     FormsModule,
   ],
   providers: [
+    MessageService,
     SupaService,
     StandupService,
     StoryService,

src/app/channel/channel.component.html

@@ -17,7 +17,7 @@
         <div class="row messageRow">
             <div class="col-12 p-0">
                 <ul class="list-group list-group-flush">
-                    <li *ngFor="let m of messages" class="list-group-item">{{m.message}}</li>
+                    <li *ngFor="let m of messages" class="list-group-item"><small class="text-muted mr-1">{{ m.created_at | date : 'HH:MM' }}</small><strong class="mr-1">{{ m.user ? m.user.username : '...' }}</strong>{{m.message}}</li>
                 </ul>
             </div>
         </div>
@@ -26,7 +26,7 @@
                 <div class="card w-100 d-flex flex-row">
                     <input [(ngModel)]="messageInput" type="text" class="input flex-grow-1">
                     <div class="optionRow">
-                        <button class="btn btn-sm btn-primary btnSend ml-1">Send</button>
+                        <button class="btn btn-sm btn-primary btnSend ml-1" (click)="sendMessage(messageInput)">Send</button>
                     </div>
                 </div>
         </div>

src/app/channel/channel.component.ts

@@ -2,11 +2,13 @@ import { Component, OnInit } from "@angular/core";
 import { DomSanitizer } from '@angular/platform-browser';
 import { ActivatedRoute, Router } from '@angular/router';
 import { NgbModal } from "@ng-bootstrap/ng-bootstrap";
+import { SupabaseAuthUser } from '@supabase/supabase-js';
 import { Subject } from 'rxjs';
-import { take, takeUntil } from 'rxjs/operators';
+import { take, takeUntil, map } from 'rxjs/operators';
 import { Channel } from '../api/supabase/channel';
 import { ChannelService } from '../api/supabase/channel.service';
 import { Message } from '../api/supabase/message';
+import { MessageService } from '../api/supabase/message.service';
 import { SupaService } from '../api/supabase/supa.service';
 import { User } from '../api/supabase/user';
 import { UserService } from '../api/supabase/user.service';
@@ -32,6 +34,7 @@ export class ChannelComponent implements OnInit {
     private sanitizer: DomSanitizer,
     private userService: UserService,
     private router: Router,
+    private messageService: MessageService,
    ) { }
 
   ngOnInit() {
@@ -43,9 +46,12 @@ export class ChannelComponent implements OnInit {
             data => {
               console.log('got channel', data);
               this.channel = data;
-              // this.loadMessages(data.id).subscribe(messages => {
-              //   this.messages = messages;
-              // });
+              this.loadMessages(data.id).subscribe(messages => {
+                this.messages = messages;
+              });
+              // for (let i = 0; i < 100; i++) {
+              //   this.messages.push(new Message(params.id, 'demo', 'hola'))
+              // }
             },
             error => {
               console.error(error);
@@ -74,12 +80,30 @@ export class ChannelComponent implements OnInit {
     return this.channelService.getOne(id).pipe(takeUntil(this.$destroy.asObservable()));
   }
 
-  loadMessages(channel_id: number) {
-
+  loadMessages(channel_id: number = this.channel.id) {
+    return this.messageService.getMessages(channel_id).pipe(
+      takeUntil(this.$destroy.asObservable()),
+      map((messages: Message[]) => {
+        messages.forEach((message: Message) => this.userService.getOne(message.user_id).subscribe(user => message.user = user));
+        return messages;
+      })
+    );
   }
 
-  loadUser(user_id: string) {
-
+  async sendMessage(text: string) {
+    if (!this.channel || !text) {
+      return;
+    }
+    const user: SupabaseAuthUser = await this.supaService.client.auth.user();
+    const message = new Message(this.channel.id, user.id, text);
+    this.messageService.addOne(message).pipe(take(1)).subscribe(
+      data => {
+        console.log('Success', data);
+      },
+      error => {
+        console.error('Failed', error);
+      }
+    );
   }
 
   updateChannel(name:string, desc:string) {

src/app/huddle/huddle.component.ts

@@ -159,10 +159,8 @@ export class HuddleComponent implements OnInit, OnDestroy {
   loadStories(id: number) {
     return this.storyService.getStories(id).pipe(
       takeUntil(this.unsubscribe.asObservable()),
-      map(stories => {
-        stories.forEach(story => {
-          this.userService.getOne(story.user_id).subscribe(user => story.user = user)
-        })
+      map((stories: Story[]) => {
+        stories.forEach((story: Story) => this.userService.getOne(story.user_id).subscribe(user => story.user = user));
         return stories;
       })
     );