EP.35 Storing Chat History in WebSocket and GraphQL Subscription

Why Store Chat History?

Storing chat history allows users to retrieve old messages, view message history, and operate the chat system continuously without losing data when the application is closed. It also enables the system to display past messages, such as:

• Fetching recent messages when a user joins a conversation
• Loading historical chat messages
• Saving messages to the database for later retrieval

Structure of the Chat History Storage System

1. WebSocket Server - Receives and sends messages via WebSocket
2. GraphQL API - Manages message sending and receiving, and stores data
3. Database (PostgreSQL / MongoDB) - Used to store chat history
4. GraphQL Subscription - Sends new messages to users in real-time

Install Necessary Libraries

go get github.com/99designs/gqlgen
go get github.com/gorilla/websocket
go get github.com/jmoiron/sqlx

Database Setup for Storing Chat History

File: schema.sql

CREATE TABLE messages (
    id SERIAL PRIMARY KEY,
    content TEXT NOT NULL,
    sender TEXT NOT NULL,
    timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

Creating GraphQL Schema for Storing Chat History

File: schema.graphql

type Query {
  messages: [Message!]!
}

type Mutation {
  sendMessage(content: String!, sender: String!): Message!
}

type Subscription {
  messageAdded: Message!
}

type Message {
  id: ID!
  content: String!
  sender: String!
  timestamp: String!
}

Creating Resolver to Connect to the Database

File: resolver.go

package main

import (
    "context"
    "database/sql"
    "fmt"
    "sync"
    "time"

    _ "github.com/lib/pq"
)

type Message struct {
    ID        int       `json:"id"`
    Content   string    `json:"content"`
    Sender    string    `json:"sender"`
    Timestamp time.Time `json:"timestamp"`
}

type Resolver struct {
    mu          sync.Mutex
    db          *sql.DB
    subscribers map[string]chan Message
}

func NewResolver(db *sql.DB) *Resolver {
    return &Resolver{
        db:          db,
        subscribers: make(map[string]chan Message),
    }
}

func (r *Resolver) Query_messages(ctx context.Context) ([]Message, error) {
    rows, err := r.db.Query("SELECT id, content, sender, timestamp FROM messages ORDER BY timestamp DESC LIMIT 50")
    if err != nil {
        return nil, err
    }
    defer rows.Close()

    var messages []Message
    for rows.Next() {
        var msg Message
        if err := rows.Scan(&msg.ID, &msg.Content, &msg.Sender, &msg.Timestamp); err != nil {
            return nil, err
        }
        messages = append(messages, msg)
    }
    return messages, nil
}

func (r *Resolver) Mutation_sendMessage(ctx context.Context, content string, sender string) (Message, error) {
    var message Message
    err := r.db.QueryRow("INSERT INTO messages (content, sender) VALUES ($1, $2) RETURNING id, content, sender, timestamp", content, sender).
        Scan(&message.ID, &message.Content, &message.Sender, &message.Timestamp)
    if err != nil {
        return Message{}, err
    }

    r.mu.Lock()
    for _, ch := range r.subscribers {
        ch <- message
    }
    r.mu.Unlock()
    return message, nil
}

func (r *Resolver) Subscription_messageAdded(ctx context.Context) (<-chan Message, error) {
    id := fmt.Sprintf("%d", len(r.subscribers))
    r.mu.Lock()
    ch := make(chan Message)
    r.subscribers[id] = ch
    r.mu.Unlock()

    go func() {
        <-ctx.Done()
        r.mu.Lock()
        delete(r.subscribers, id)
        close(ch)
        r.mu.Unlock()
    }()
    return ch, nil
}

Creating WebSocket Server and Connecting to the Database

File: server.go

package main

import (
    "database/sql"
    "log"
    "net/http"
    "github.com/99designs/gqlgen/graphql/handler"
    "github.com/99designs/gqlgen/graphql/playground"
)

func main() {
    db, err := sql.Open("postgres", "user=postgres password=secret dbname=chat sslmode=disable")
    if err != nil {
        log.Fatal(err)
    }
    defer db.Close()

    resolver := NewResolver(db)
    srv := handler.NewDefaultServer(NewExecutableSchema(Config{Resolvers: resolver}))

    http.Handle("/", playground.Handler("GraphQL Playground", "/query"))
    http.Handle("/query", srv)

    log.Println("Chat Server with Database running at http://localhost:8080")
    log.Fatal(http.ListenAndServe(":8080", nil))
}

Challenge!

Try adding features for searching old messages and filtering messages by sender in your chat system.

Next EP

In EP.36, we will add User Authentication to the WebSocket Chat to control access permissions 🚀