feat: Stream connectivity status (#36)

* chore: Remove double exlamation mark

* feat: 1st approach of getting connected peers

* fix: Include setup test in jest config

* feat: Scheduler adapter to store connected peers in redis

* feat: Memory cache or Redis

* feat: Get Friends filter by connectivity status using Redis

* refactor: Move component interfaces to root types file

* chore: Remove unused imports

* feat: GetFriends does not filter by connectivity status

* test: Adapt tests

* feat: Send id of the request to subscribe, get sent and get pending

* feat: Peer Tracking + Subscribe to Friends Status Updates

* test: Peer tracking tests

* refactor: Rename of one of the base components

* refactor: Reuse common queries

* test: Archipelago Stats tests

* test: Friend Updates Streamer tests

* test: PubSub tests

* style: Missing eof

* chore: Add TODO

* fix: Propagate error if the fetch to stats fails

* refactor: Rename confusing db function to filterActiveFriendshipsFromAddresses

* test: New friendship function test

* chore: First README iteration

* chore: Update DER

* chore: Improve seq diagram

* chore: More readme improvements

* chore: Some fixes in the seq diag

* chore: Add TODO comment

* test: Tiny improvements on tests for more coverage

.env.default

@@ -13,3 +13,4 @@ HTTP_SERVER_HOST=0.0.0.0
 # reset metrics at 00:00UTC
 WKC_METRICS_RESET_AT_NIGHT=false
 
+NATS_URL=localhost:4222

README.md

@@ -1,3 +1,232 @@
 # EA Social Service
 
 [![Coverage Status](https://coveralls.io/repos/github/decentraland/social-service-ea/badge.svg)](https://coveralls.io/github/decentraland/social-service-ea)
+
+A microservice that handles social interactions (friendships) for Decentraland, built using the Well Known Components architecture pattern.
+
+## Table of Contents
+
+- [🌟 Features](#-features)
+- [🏗 Architecture](#-architecture)
+  - [Component-Based Architecture](#component-based-architecture)
+  - [Database Design](#database-design)
+  - [Flow Diagrams](#flow-diagrams)
+- [🚀 Getting Started](#-getting-started)
+  - [Prerequisites](#prerequisites)
+  - [Local Development](#local-development)
+  - [Environment Variables](#environment-variables)
+- [🧪 Testing](#-testing)
+  - [Test Coverage](#test-coverage)
+- [🔄 CI/CD](#-ci/cd)
+  - [Deployment Environments](#deployment-environments)
+
+## 🌟 Features
+
+- Friendship management (requests, accepts, rejects, cancellations)
+- Real-time friend status updates
+- Mutual friends discovery
+- Online status tracking
+- Integration with Archipelago for peer synchronization
+
+## 🏗 Architecture
+
+### Component-Based Architecture
+
+This service follows the Well Known Components pattern, where each component is a self-contained unit with a clear interface. The main components are:
+
+- **Database (PostgreSQL)**: Stores friendship relationships and actions
+- **Cache (Redis)**: Handles temporary data and real-time status
+- **RPC Server**: Manages client-server RPC communication
+- **PubSub**: Handles real-time updates
+- **Archipelago Stats**: Integrates with Decentraland's peer discovery system
+- **Peer Tracking**: Monitors online status of users through the NATS messaging system
+- **Peers Synchronization**: Synchronizes peers with the Archipelago Stats service and store them in Redis
+
+### Database Design
+
+```plantuml
+@startuml
+!define table(x) class x << (T,#FFAAAA) >>
+!define primary_key(x) <u>x</u>
+!define foreign_key(x) #x#
+hide methods
+hide stereotypes
+
+table(friendships) {
+  primary_key(id): uuid
+  address_requester: varchar
+  address_requested: varchar
+  is_active: boolean
+  created_at: timestamp
+  updated_at: timestamp
+  --
+  indexes
+  ..
+  hash(address_requester)
+  hash(address_requested)
+  btree(LOWER(address_requester))
+  btree(LOWER(address_requested))
+}
+
+table(friendship_actions) {
+  primary_key(id): uuid
+  foreign_key(friendship_id): uuid
+  action: varchar
+  acting_user: varchar
+  metadata: jsonb
+  timestamp: timestamp
+  --
+  indexes
+  ..
+  btree(friendship_id)
+}
+
+friendships ||--|{ friendship_actions
+@enduml
+```
+
+The database schema supports:
+
+- Bidirectional friendships
+- Action history tracking
+- Metadata for requests
+- Optimized queries with proper indexes
+
+See migrations for details: [migrations](./src/migrations)
+
+### Flow Diagrams
+
+```mermaid
+sequenceDiagram
+    participant Client
+    participant WebSocket
+    participant RPC Server
+    participant Redis
+    participant NATS
+    participant DB
+
+    Note over Client,DB: Connection Setup
+    Client->>WebSocket: WS Handshake
+    activate WebSocket
+    WebSocket-->>Client: Connection Established
+    Client->>WebSocket: Auth Message
+    WebSocket->>RPC Server: Attach Transport
+    activate RPC Server
+
+    Note over RPC Server,NATS: Subscriptions Setup
+    RPC Server->>Redis: Subscribe to updates channels
+    activate Redis
+    Note over Redis: friendship.updates
+    Note over Redis: friend.status.updates
+    RPC Server->>NATS: Subscribe to peer events
+    activate NATS
+    Note over NATS: peer.*.connected
+    Note over NATS: peer.*.disconnected
+    Note over NATS: peer.*.heartbeat
+
+    Note over Client,DB: Friendship Request Flow
+    Client->>RPC Server: Friend Request
+    RPC Server->>DB: Create Friendship Record
+    DB-->>RPC Server: Friendship Created
+    RPC Server->>DB: Record Friendship Action
+    RPC Server->>Redis: Publish Friendship Update
+    RPC Server-->>Client: Request Confirmation
+
+    Note over Client,DB: Friendship Updates Flow
+    Redis-->>RPC Server: Friendship Update Event
+    RPC Server-->>Client: Stream Friendship Updates
+    Note over RPC Server: (accept/cancel/reject/delete)
+
+    Note over Client,DB: Friends Lifecycle
+    NATS-->>Redis: Peer Heartbeat
+    Redis-->>RPC Server: Friend Status Update
+    RPC Server->>Redis: Request Cached Peers
+    Redis-->>RPC Server: Cached Peers
+    RPC Server->>DB: Request Online Friends
+    DB-->>RPC Server: Online Friends
+    RPC Server->>DB: Query Online Friends
+    RPC Server-->>Client: Stream Friend Status Updates
+    Note over RPC Server: (online/offline)
+
+    Note over Client,DB: Cleanup
+    Client->>WebSocket: Connection Close
+    WebSocket->>RPC Server: Detach Transport
+    RPC Server->>Redis: Unsubscribe
+    RPC Server->>NATS: Unsubscribe
+    deactivate WebSocket
+    deactivate RPC Server
+    deactivate Redis
+    deactivate NATS
+```
+
+## 🚀 Getting Started
+
+### Prerequisites
+
+- Node.js v18.20.4
+- Docker and Docker Compose
+- PostgreSQL
+- Redis
+
+### Local Development
+
+1. Clone the repository
+2. Install dependencies:
+
+```bash
+yarn install
+```
+
+3. Start the development environment:
+
+```bash
+docker-compose up -d
+```
+
+4. Run migrations:
+
+```bash
+yarn migrate up
+```
+
+5. Run the service:
+
+```bash
+yarn start
+```
+
+### Environment Variables
+
+Key environment variables needed:
+
+- `REDIS_HOST`: URL of the Redis instance
+- `PG_COMPONENT_PSQL_CONNECTION_STRING`: URL of the PostgreSQL instance
+- `ARCHIPELAGO_STATS_URL`: URL of the Archipelago Stats service
+
+See `.env.default` for all available options.
+
+## 🧪 Testing
+
+The project uses Jest for testing. Run tests with:
+
+```bash
+yarn test
+```
+
+### Test Coverage
+
+Coverage reports are generated in the `/coverage` directory and uploaded to Coveralls.
+
+## 🔄 CI/CD
+
+The project uses GitHub Actions for:
+
+- Continuous Integration
+- Docker image building
+- Automated deployments to dev/prod environments
+- Dependency management with Dependabot
+
+### Deployment Environments
+
+- **Development**: Automatic deployments on main branch
+- **Production**: Manual deployments via GitHub releases

jest.config.js

@@ -8,5 +8,6 @@ module.exports = {
   coverageDirectory: 'coverage',
   collectCoverageFrom: ['src/**/*.ts', 'src/**/*.js', '!src/migrations/**'],
   testMatch: ['**/*.spec.(ts)'],
-  testEnvironment: 'node'
+  testEnvironment: 'node',
+  setupFilesAfterEnv: ['<rootDir>/test/setupTests.ts']
 }

package.json

@@ -30,17 +30,19 @@
   },
   "dependencies": {
     "@dcl/platform-crypto-middleware": "^1.1.0",
-    "@dcl/protocol": "^1.0.0-12815643167.commit-c4162c4",
+    "@dcl/protocol": "https://sdk-team-cdn.decentraland.org/@dcl/protocol/branch//dcl-protocol-1.0.0-12890706635.commit-a7e4210.tgz",
     "@dcl/rpc": "^1.1.2",
     "@well-known-components/env-config-provider": "^1.2.0",
     "@well-known-components/fetch-component": "^2.0.2",
     "@well-known-components/http-server": "^2.1.0",
     "@well-known-components/interfaces": "^1.4.3",
     "@well-known-components/logger": "^3.1.3",
     "@well-known-components/metrics": "^2.1.0",
+    "@well-known-components/nats-component": "^2.0.0",
     "@well-known-components/pg-component": "^0.2.2",
     "@well-known-components/uws-http-server": "^0.0.2",
     "fp-future": "^1.0.1",
+    "lru-cache": "^10.4.3",
     "mitt": "^3.0.1",
     "redis": "^4.6.13",
     "sql-template-strings": "^2.2.2",

src/adapters/archipelago-stats.ts

@@ -0,0 +1,34 @@
+import { AppComponents, IArchipelagoStatsComponent } from '../types'
+import { PEERS_CACHE_KEY } from '../utils/peers'
+
+export async function createArchipelagoStatsComponent({
+  logs,
+  config,
+  fetcher,
+  redis
+}: Pick<AppComponents, 'logs' | 'config' | 'fetcher' | 'redis'>): Promise<IArchipelagoStatsComponent> {
+  const logger = logs.getLogger('archipelago-stats-component')
+  const url = await config.getString('ARCHIPELAGO_STATS_URL')
+
+  return {
+    async getPeers() {
+      try {
+        const response = await fetcher.fetch(`${url}/comms/peers`)
+
+        if (!response.ok) {
+          throw new Error(`Error fetching peers: ${response.statusText}`)
+        }
+
+        const { peers } = await response.json()
+
+        return peers.map((peer: { id: string }) => peer.id)
+      } catch (error: any) {
+        logger.error(`Error fetching peers from archipelago stats: ${error.message}`)
+        throw error
+      }
+    },
+    async getPeersFromCache() {
+      return (await redis.get<string[]>(PEERS_CACHE_KEY)) || []
+    }
+  }
+}