HyperSpaceOut/IRC-kosmi-relay
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

minor tweak

Browse Source
This commit is contained in:
cottongin
2025-10-31 21:41:27 -04:00
parent c63c29efc9
commit 3116721256
7 changed files with 14 additions and 2000 deletions
Download Patch File Download Diff File Expand all files Collapse all files

30
Dockerfile
View File

@@ -1,28 +1,10 @@
# Single-stage build for Matterbridge with Kosmi bridge (Playwright) # Single-stage build for Matterbridge with Kosmi bridge (Native WebSocket)
FROM golang:1.23-bookworm FROM golang:1.23-alpine
WORKDIR /app WORKDIR /app
# Install system dependencies for Playwright Chromium # Install only essential dependencies
RUN apt-get update && apt-get install -y \ RUN apk add --no-cache ca-certificates
ca-certificates \
chromium \
libnss3 \
libnspr4 \
libatk1.0-0 \
libatk-bridge2.0-0 \
libcups2 \
libdrm2 \
libdbus-1-3 \
libxkbcommon0 \
libxcomposite1 \
libxdamage1 \
libxfixes3 \
libxrandr2 \
libgbm1 \
libasound2 \
libatspi2.0-0 \
&& rm -rf /var/lib/apt/lists/*
# Copy go mod files # Copy go mod files
COPY go.mod go.sum ./ COPY go.mod go.sum ./
@@ -34,10 +16,6 @@ COPY . .
# Build matterbridge # Build matterbridge
RUN go build -o matterbridge . RUN go build -o matterbridge .
# Install playwright-go CLI and drivers
RUN go install github.com/playwright-community/playwright-go/cmd/playwright@latest && \
$(go env GOPATH)/bin/playwright install --with-deps chromium
# Copy configuration # Copy configuration
COPY matterbridge.toml /app/matterbridge.toml.example COPY matterbridge.toml /app/matterbridge.toml.example

612
bridge/kosmi/chromedp_client.go
View File

@@ -1,612 +0,0 @@
package bkosmi
import (
"context"
"encoding/json"
"fmt"
"strings"
"sync"
"time"
"github.com/chromedp/cdproto/input"
"github.com/chromedp/cdproto/page"
"github.com/chromedp/chromedp"
"github.com/sirupsen/logrus"
)
// ChromeDPClient manages a headless Chrome instance to connect to Kosmi
type ChromeDPClient struct {
ctx context.Context
cancel context.CancelFunc
roomURL string
log *logrus.Entry
messageHandlers []func(*NewMessagePayload)
mu sync.RWMutex
connected bool
}
// NewChromeDPClient creates a new ChromeDP-based Kosmi client
func NewChromeDPClient(roomURL string, log *logrus.Entry) *ChromeDPClient {
return &ChromeDPClient{
roomURL: roomURL,
log: log,
messageHandlers: []func(*NewMessagePayload){},
}
}
// Connect launches Chrome and navigates to the Kosmi room
func (c *ChromeDPClient) Connect() error {
c.log.Info("Launching headless Chrome for Kosmi connection")
// Create Chrome context with flags to avoid headless detection
opts := append(chromedp.DefaultExecAllocatorOptions[:],
chromedp.Flag("headless", true),
chromedp.Flag("disable-gpu", false), // Enable GPU to look more real
chromedp.Flag("no-sandbox", true),
chromedp.Flag("disable-dev-shm-usage", true),
chromedp.Flag("disable-blink-features", "AutomationControlled"), // Hide automation
chromedp.Flag("disable-infobars", true),
chromedp.Flag("window-size", "1920,1080"),
chromedp.UserAgent("Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"),
)
allocCtx, allocCancel := chromedp.NewExecAllocator(context.Background(), opts...)
ctx, cancel := chromedp.NewContext(allocCtx)
c.ctx = ctx
c.cancel = func() {
cancel()
allocCancel()
}
// Inject anti-detection scripts and WebSocket hook BEFORE any navigation
c.log.Info("Injecting anti-detection and WebSocket interceptor...")
if err := c.injectAntiDetection(); err != nil {
return fmt.Errorf("failed to inject anti-detection: %w", err)
}
if err := c.injectWebSocketHookBeforeLoad(); err != nil {
return fmt.Errorf("failed to inject WebSocket hook: %w", err)
}
// Now navigate to the room
c.log.Infof("Navigating to Kosmi room: %s", c.roomURL)
if err := chromedp.Run(ctx,
chromedp.Navigate(c.roomURL),
chromedp.WaitReady("body"),
); err != nil {
return fmt.Errorf("failed to navigate to room: %w", err)
}
c.log.Info("Page loaded, checking if hook is active...")
// Verify the hook is installed
var hookInstalled bool
if err := chromedp.Run(ctx, chromedp.Evaluate(`window.__KOSMI_WS_HOOK_INSTALLED__ === true`, &hookInstalled)); err != nil {
c.log.Warnf("Could not verify hook installation: %v", err)
} else if hookInstalled {
c.log.Info("✓ WebSocket hook confirmed installed")
} else {
c.log.Warn("✗ WebSocket hook not detected!")
}
// Wait a moment for WebSocket to connect
c.log.Info("Waiting for WebSocket connection...")
time.Sleep(3 * time.Second)
// Check if we've captured any WebSocket connections
var wsConnected string
checkScript := `
(function() {
if (window.__KOSMI_WS_CONNECTED__) {
return 'WebSocket connection intercepted';
}
return 'No WebSocket connection detected yet';
})();
`
if err := chromedp.Run(ctx, chromedp.Evaluate(checkScript, &wsConnected)); err == nil {
c.log.Infof("Status: %s", wsConnected)
}
c.mu.Lock()
c.connected = true
c.mu.Unlock()
c.log.Info("Successfully connected to Kosmi via Chrome")
// Start console log listener (for debugging)
go c.listenToConsole()
// Start message listener
go c.listenForMessages()
return nil
}
// injectAntiDetection injects scripts to hide automation/headless detection
func (c *ChromeDPClient) injectAntiDetection() error {
script := `
// Override navigator.webdriver
Object.defineProperty(navigator, 'webdriver', {
get: () => false,
});
// Override plugins
Object.defineProperty(navigator, 'plugins', {
get: () => [1, 2, 3, 4, 5],
});
// Override languages
Object.defineProperty(navigator, 'languages', {
get: () => ['en-US', 'en'],
});
// Chrome runtime
window.chrome = {
runtime: {},
};
// Permissions
const originalQuery = window.navigator.permissions.query;
window.navigator.permissions.query = (parameters) => (
parameters.name === 'notifications' ?
Promise.resolve({ state: Notification.permission }) :
originalQuery(parameters)
);
console.log('[Kosmi Bridge] Anti-detection scripts injected');
`
return chromedp.Run(c.ctx, chromedp.ActionFunc(func(ctx context.Context) error {
_, err := page.AddScriptToEvaluateOnNewDocument(script).Do(ctx)
return err
}))
}
// injectWebSocketHookBeforeLoad uses CDP to inject script before any page scripts run
func (c *ChromeDPClient) injectWebSocketHookBeforeLoad() error {
// Get the WebSocket hook script
script := c.getWebSocketHookScript()
// Use chromedp.ActionFunc to access the CDP directly
return chromedp.Run(c.ctx, chromedp.ActionFunc(func(ctx context.Context) error {
// Use Page.addScriptToEvaluateOnNewDocument to inject before page load
// This is the proper way to inject scripts that run before page JavaScript
_, err := page.AddScriptToEvaluateOnNewDocument(script).Do(ctx)
return err
}))
}
// getWebSocketHookScript returns the WebSocket interception script
func (c *ChromeDPClient) getWebSocketHookScript() string {
return `
(function() {
if (window.__KOSMI_WS_HOOK_INSTALLED__) {
console.log('[Kosmi Bridge] WebSocket hook already installed');
return;
}
// Store original WebSocket constructor
const OriginalWebSocket = window.WebSocket;
// Store messages in a queue
window.__KOSMI_MESSAGE_QUEUE__ = [];
// Hook WebSocket constructor
window.WebSocket = function(url, protocols) {
const socket = new OriginalWebSocket(url, protocols);
// Check if this is Kosmi's GraphQL WebSocket
if (url.includes('engine.kosmi.io') || url.includes('gql-ws')) {
console.log('[Kosmi Bridge] WebSocket hook active for:', url);
window.__KOSMI_WS_CONNECTED__ = true;
// Method 1: Hook addEventListener
const originalAddEventListener = socket.addEventListener.bind(socket);
socket.addEventListener = function(type, listener, options) {
if (type === 'message') {
const wrappedListener = function(event) {
try {
const data = JSON.parse(event.data);
console.log('[Kosmi Bridge] Message intercepted:', data);
window.__KOSMI_MESSAGE_QUEUE__.push({
timestamp: Date.now(),
data: data,
source: 'addEventListener'
});
} catch (e) {
// Not JSON, skip
}
// Call original listener
return listener.call(this, event);
};
return originalAddEventListener(type, wrappedListener, options);
}
return originalAddEventListener(type, listener, options);
};
// Method 2: Intercept onmessage property setter
let realOnMessage = null;
const descriptor = Object.getOwnPropertyDescriptor(WebSocket.prototype, 'onmessage');
Object.defineProperty(socket, 'onmessage', {
get: function() {
return realOnMessage;
},
set: function(handler) {
realOnMessage = function(event) {
try {
const data = JSON.parse(event.data);
console.log('[Kosmi Bridge] Message via onmessage:', data);
window.__KOSMI_MESSAGE_QUEUE__.push({
timestamp: Date.now(),
data: data,
source: 'onmessage'
});
} catch (e) {
// Not JSON, skip
}
// ALWAYS call original handler
if (handler) {
handler.call(socket, event);
}
};
// Set it on the underlying WebSocket
if (descriptor && descriptor.set) {
descriptor.set.call(socket, realOnMessage);
}
},
configurable: true
});
console.log('[Kosmi Bridge] WebSocket hooks installed');
}
return socket;
};
// Preserve the original constructor properties
window.WebSocket.prototype = OriginalWebSocket.prototype;
window.WebSocket.CONNECTING = OriginalWebSocket.CONNECTING;
window.WebSocket.OPEN = OriginalWebSocket.OPEN;
window.WebSocket.CLOSING = OriginalWebSocket.CLOSING;
window.WebSocket.CLOSED = OriginalWebSocket.CLOSED;
window.__KOSMI_WS_HOOK_INSTALLED__ = true;
console.log('[Kosmi Bridge] WebSocket hook installed successfully');
})();
`
}
// listenToConsole captures console logs from the browser
func (c *ChromeDPClient) listenToConsole() {
ticker := time.NewTicker(1 * time.Second)
defer ticker.Stop()
for {
select {
case <-c.ctx.Done():
return
case <-ticker.C:
var logs string
script := `
(function() {
if (!window.__KOSMI_CONSOLE_LOGS__) {
window.__KOSMI_CONSOLE_LOGS__ = [];
const originalLog = console.log;
const originalWarn = console.warn;
const originalError = console.error;
console.log = function(...args) {
window.__KOSMI_CONSOLE_LOGS__.push({type: 'log', message: args.join(' ')});
originalLog.apply(console, args);
};
console.warn = function(...args) {
window.__KOSMI_CONSOLE_LOGS__.push({type: 'warn', message: args.join(' ')});
originalWarn.apply(console, args);
};
console.error = function(...args) {
window.__KOSMI_CONSOLE_LOGS__.push({type: 'error', message: args.join(' ')});
originalError.apply(console, args);
};
}
const logs = window.__KOSMI_CONSOLE_LOGS__;
window.__KOSMI_CONSOLE_LOGS__ = [];
return JSON.stringify(logs);
})();
`
if err := chromedp.Run(c.ctx, chromedp.Evaluate(script, &logs)); err != nil {
continue
}
if logs != "" && logs != "[]" {
var logEntries []struct {
Type string `json:"type"`
Message string `json:"message"`
}
if err := json.Unmarshal([]byte(logs), &logEntries); err == nil {
for _, entry := range logEntries {
// Only show Kosmi Bridge logs
if strings.Contains(entry.Message, "[Kosmi Bridge]") {
switch entry.Type {
case "error":
c.log.Errorf("Browser: %s", entry.Message)
case "warn":
c.log.Warnf("Browser: %s", entry.Message)
default:
c.log.Debugf("Browser: %s", entry.Message)
}
}
}
}
}
}
}
}
// listenForMessages continuously polls for new messages from the queue
func (c *ChromeDPClient) listenForMessages() {
c.log.Info("Starting message listener")
ticker := time.NewTicker(500 * time.Millisecond) // Poll every 500ms
defer ticker.Stop()
for {
select {
case <-c.ctx.Done():
c.log.Info("Message listener stopped")
return
case <-ticker.C:
if err := c.pollMessages(); err != nil {
c.log.Errorf("Error polling messages: %v", err)
}
}
}
}
// pollMessages retrieves and processes messages from the queue
func (c *ChromeDPClient) pollMessages() error {
var messagesJSON string
// Get and clear the message queue
script := `
(function() {
const messages = window.__KOSMI_MESSAGE_QUEUE__ || [];
const count = messages.length;
window.__KOSMI_MESSAGE_QUEUE__ = [];
if (count > 0) {
console.log('[Kosmi Bridge] Polling found', count, 'messages');
}
return JSON.stringify(messages);
})();
`
if err := chromedp.Run(c.ctx, chromedp.Evaluate(script, &messagesJSON)); err != nil {
return err
}
if messagesJSON == "" || messagesJSON == "[]" {
return nil // No messages
}
c.log.Debugf("Retrieved %d bytes of message data", len(messagesJSON))
// Parse messages
var messages []struct {
Timestamp int64 `json:"timestamp"`
Data json.RawMessage `json:"data"`
Source string `json:"source"`
}
if err := json.Unmarshal([]byte(messagesJSON), &messages); err != nil {
c.log.Errorf("Failed to parse messages JSON: %v", err)
c.log.Debugf("Raw JSON: %s", messagesJSON)
return fmt.Errorf("failed to parse messages: %w", err)
}
c.log.Infof("Processing %d messages from queue", len(messages))
// Process each message
for i, msg := range messages {
c.log.Debugf("Processing message %d/%d from source: %s", i+1, len(messages), msg.Source)
c.processMessage(msg.Data)
}
return nil
}
// processMessage handles a single GraphQL message
func (c *ChromeDPClient) processMessage(data json.RawMessage) {
// Parse as GraphQL message
var gqlMsg struct {
Type string `json:"type"`
Payload json.RawMessage `json:"payload"`
}
if err := json.Unmarshal(data, &gqlMsg); err != nil {
c.log.Debugf("Failed to parse GraphQL message: %v", err)
return
}
// Only process "next" or "data" type messages
if gqlMsg.Type != "next" && gqlMsg.Type != "data" {
return
}
// Parse the payload
var payload NewMessagePayload
if err := json.Unmarshal(gqlMsg.Payload, &payload); err != nil {
c.log.Debugf("Failed to parse message payload: %v", err)
return
}
// Check if this is a newMessage event
if payload.Data.NewMessage.Body == "" {
return // Not a message event
}
// Call all registered handlers
c.mu.RLock()
handlers := c.messageHandlers
c.mu.RUnlock()
for _, handler := range handlers {
handler(&payload)
}
}
// OnMessage registers a handler for incoming messages
func (c *ChromeDPClient) OnMessage(handler func(*NewMessagePayload)) {
c.mu.Lock()
defer c.mu.Unlock()
c.messageHandlers = append(c.messageHandlers, handler)
}
// SendMessage sends a message to the Kosmi room
func (c *ChromeDPClient) SendMessage(text string) error {
c.mu.RLock()
if !c.connected {
c.mu.RUnlock()
return fmt.Errorf("not connected")
}
c.mu.RUnlock()
// Wait for the chat input to be available (with timeout)
// Kosmi uses a contenteditable div with role="textbox"
var inputFound bool
for i := 0; i < 50; i++ {
// Simple check without string replacement issues
checkScript := `
(function() {
const input = document.querySelector('div[role="textbox"][contenteditable="true"]');
return input !== null;
})();
`
if err := chromedp.Run(c.ctx, chromedp.Evaluate(checkScript, &inputFound)); err != nil {
return fmt.Errorf("failed to check for chat input: %w", err)
}
if inputFound {
c.log.Infof("Chat input found after %d attempts (%.1f seconds)", i, float64(i)*0.1)
break
}
// Log progress periodically
if i == 0 || i == 10 || i == 25 || i == 49 {
c.log.Debugf("Still waiting for chat input... attempt %d/50", i+1)
}
time.Sleep(100 * time.Millisecond)
}
if !inputFound {
// Diagnostic: Get element counts directly
var diagInfo struct {
Textareas int `json:"textareas"`
Contenteditable int `json:"contenteditable"`
Textboxes int `json:"textboxes"`
}
diagScript := `
(function() {
return {
textareas: document.querySelectorAll('textarea').length,
contenteditable: document.querySelectorAll('[contenteditable="true"]').length,
textboxes: document.querySelectorAll('[role="textbox"]').length
};
})();
`
if err := chromedp.Run(c.ctx, chromedp.Evaluate(diagScript, &diagInfo)); err == nil {
c.log.Errorf("Diagnostic: textareas=%d, contenteditable=%d, textboxes=%d",
diagInfo.Textareas, diagInfo.Contenteditable, diagInfo.Textboxes)
}
c.log.Error("Chat input not found after 5 seconds")
return fmt.Errorf("chat input not available after timeout")
}
// Send the message using ChromeDP's native SendKeys
// This is more reliable than dispatching events manually and mimics actual user input
selector := `div[role="textbox"][contenteditable="true"]`
// First, clear the input and type the message
err := chromedp.Run(c.ctx,
chromedp.Focus(selector),
chromedp.Evaluate(`document.querySelector('div[role="textbox"][contenteditable="true"]').textContent = ''`, nil),
chromedp.SendKeys(selector, text),
)
if err != nil {
return fmt.Errorf("failed to type message: %w", err)
}
// Small delay for React to process
time.Sleep(100 * time.Millisecond)
// Send Enter key using CDP Input API directly
// This is closer to what Playwright does and should trigger all the right events
err = chromedp.Run(c.ctx,
chromedp.ActionFunc(func(ctx context.Context) error {
// Send keyDown for Enter
if err := input.DispatchKeyEvent(input.KeyDown).
WithKey("Enter").
WithCode("Enter").
WithNativeVirtualKeyCode(13).
WithWindowsVirtualKeyCode(13).
Do(ctx); err != nil {
return err
}
// Send keyUp for Enter
return input.DispatchKeyEvent(input.KeyUp).
WithKey("Enter").
WithCode("Enter").
WithNativeVirtualKeyCode(13).
WithWindowsVirtualKeyCode(13).
Do(ctx)
}),
)
if err != nil {
return fmt.Errorf("failed to send message via SendKeys: %w", err)
}
c.log.Debugf("Sent message: %s", text)
return nil
}
// Close closes the Chrome instance
func (c *ChromeDPClient) Close() error {
c.log.Info("Closing ChromeDP client")
c.mu.Lock()
c.connected = false
c.mu.Unlock()
if c.cancel != nil {
c.cancel()
}
return nil
}
// IsConnected returns whether the client is connected
func (c *ChromeDPClient) IsConnected() bool {
c.mu.RLock()
defer c.mu.RUnlock()
return c.connected
}
// escapeJSString escapes a string for use in JavaScript
func escapeJSString(s string) string {
b, _ := json.Marshal(s)
return string(b)
}

481
bridge/kosmi/hybrid_client.go
View File

@@ -1,481 +0,0 @@
package bkosmi
import (
"context"
"encoding/json"
"fmt"
"net/http"
"net/http/cookiejar"
"net/url"
"strings"
"sync"
"time"
"github.com/chromedp/cdproto/page"
"github.com/chromedp/chromedp"
"github.com/sirupsen/logrus"
)
// HybridClient uses ChromeDP for auth/cookies and GraphQL for sending messages
type HybridClient struct {
roomURL string
roomID string
log *logrus.Entry
ctx context.Context
cancel context.CancelFunc
httpClient *http.Client
messageCallback func(*NewMessagePayload)
connected bool
mu sync.RWMutex
}
// NewHybridClient creates a new hybrid client
func NewHybridClient(roomURL string, log *logrus.Entry) *HybridClient {
return &HybridClient{
roomURL: roomURL,
log: log,
}
}
// Connect launches Chrome, gets cookies, and sets up GraphQL client
func (c *HybridClient) Connect() error {
c.log.Info("Launching Chrome to obtain session cookies")
// Extract room ID
roomID, err := extractRoomID(c.roomURL)
if err != nil {
return fmt.Errorf("failed to extract room ID: %w", err)
}
c.roomID = roomID
// Create Chrome context with anti-detection
opts := append(chromedp.DefaultExecAllocatorOptions[:],
chromedp.Flag("headless", true),
chromedp.Flag("disable-gpu", false),
chromedp.Flag("no-sandbox", true),
chromedp.Flag("disable-dev-shm-usage", true),
chromedp.Flag("disable-blink-features", "AutomationControlled"),
chromedp.Flag("disable-infobars", true),
chromedp.Flag("window-size", "1920,1080"),
chromedp.UserAgent("Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"),
)
allocCtx, allocCancel := chromedp.NewExecAllocator(context.Background(), opts...)
ctx, cancel := chromedp.NewContext(allocCtx)
c.ctx = ctx
c.cancel = func() {
cancel()
allocCancel()
}
// Inject scripts to run on every new document BEFORE creating any pages
// This ensures they run BEFORE any page JavaScript
c.log.Info("Injecting scripts to run on every page load...")
antiDetectionScript := `
Object.defineProperty(navigator, 'webdriver', { get: () => false });
Object.defineProperty(navigator, 'plugins', { get: () => [1, 2, 3, 4, 5] });
Object.defineProperty(navigator, 'languages', { get: () => ['en-US', 'en'] });
window.chrome = { runtime: {} };
`
wsHookScript := c.getWebSocketHookScript()
// Use Page.addScriptToEvaluateOnNewDocument via CDP
if err := chromedp.Run(ctx,
chromedp.ActionFunc(func(ctx context.Context) error {
_, err := page.AddScriptToEvaluateOnNewDocument(antiDetectionScript).Do(ctx)
if err != nil {
return fmt.Errorf("failed to add anti-detection script: %w", err)
}
_, err = page.AddScriptToEvaluateOnNewDocument(wsHookScript).Do(ctx)
if err != nil {
return fmt.Errorf("failed to add WebSocket hook script: %w", err)
}
return nil
}),
); err != nil {
return fmt.Errorf("failed to inject scripts: %w", err)
}
// Now navigate to the room - scripts will run before page JS
c.log.Infof("Navigating to Kosmi room: %s", c.roomURL)
if err := chromedp.Run(ctx,
chromedp.Navigate(c.roomURL),
chromedp.WaitReady("body"),
); err != nil {
return fmt.Errorf("failed to navigate to room: %w", err)
}
// Wait for page to load and WebSocket to connect
c.log.Info("Waiting for page to load and WebSocket to connect...")
time.Sleep(5 * time.Second)
// Check if WebSocket is connected
var wsStatus map[string]interface{}
checkScript := `
(function() {
return {
hookInstalled: !!window.__KOSMI_WS_HOOK_INSTALLED__,
wsFound: !!window.__KOSMI_WS__,
wsConnected: window.__KOSMI_WS__ ? window.__KOSMI_WS__.readyState === WebSocket.OPEN : false,
wsState: window.__KOSMI_WS__ ? window.__KOSMI_WS__.readyState : -1
};
})();
`
if err := chromedp.Run(ctx, chromedp.Evaluate(checkScript, &wsStatus)); err == nil {
c.log.Infof("WebSocket status: %+v", wsStatus)
}
// Get cookies from the browser
c.log.Info("Extracting cookies from browser session...")
cookies, err := c.getCookies()
if err != nil {
return fmt.Errorf("failed to get cookies: %w", err)
}
c.log.Infof("Obtained %d cookies from browser", len(cookies))
// Set up HTTP client with cookies
jar, err := cookiejar.New(nil)
if err != nil {
return fmt.Errorf("failed to create cookie jar: %w", err)
}
c.httpClient = &http.Client{
Jar: jar,
Timeout: 30 * time.Second,
}
// Add cookies to the jar
u, _ := url.Parse("https://engine.kosmi.io")
c.httpClient.Jar.SetCookies(u, cookies)
c.mu.Lock()
c.connected = true
c.mu.Unlock()
c.log.Info("Successfully connected - browser session established with cookies")
// Start message listener (using WebSocket hook in browser)
go c.listenForMessages()
return nil
}
// getCookies extracts cookies from the Chrome session
func (c *HybridClient) getCookies() ([]*http.Cookie, error) {
var cookiesData []map[string]interface{}
script := `
(function() {
return document.cookie.split(';').map(c => {
const parts = c.trim().split('=');
return {
name: parts[0],
value: parts.slice(1).join('=')
};
});
})();
`
if err := chromedp.Run(c.ctx, chromedp.Evaluate(script, &cookiesData)); err != nil {
return nil, err
}
cookies := make([]*http.Cookie, 0, len(cookiesData))
for _, cd := range cookiesData {
if name, ok := cd["name"].(string); ok {
if value, ok := cd["value"].(string); ok {
cookies = append(cookies, &http.Cookie{
Name: name,
Value: value,
})
}
}
}
return cookies, nil
}
// injectAntiDetection injects anti-detection scripts
func (c *HybridClient) injectAntiDetection() error {
script := `
Object.defineProperty(navigator, 'webdriver', { get: () => false });
Object.defineProperty(navigator, 'plugins', { get: () => [1, 2, 3, 4, 5] });
Object.defineProperty(navigator, 'languages', { get: () => ['en-US', 'en'] });
window.chrome = { runtime: {} };
`
return chromedp.Run(c.ctx, chromedp.ActionFunc(func(ctx context.Context) error {
return chromedp.Evaluate(script, nil).Do(ctx)
}))
}
// injectWebSocketHook injects the WebSocket interception script
func (c *HybridClient) injectWebSocketHook() error {
script := c.getWebSocketHookScript()
return chromedp.Run(c.ctx, chromedp.ActionFunc(func(ctx context.Context) error {
return chromedp.Evaluate(script, nil).Do(ctx)
}))
}
// getWebSocketHookScript returns the WebSocket hook JavaScript
func (c *HybridClient) getWebSocketHookScript() string {
return `
(function() {
if (window.__KOSMI_WS_HOOK_INSTALLED__) return;
const OriginalWebSocket = window.WebSocket;
window.__KOSMI_MESSAGE_QUEUE__ = [];
window.__KOSMI_WS__ = null; // Store reference to the WebSocket
window.WebSocket = function(url, protocols) {
const socket = new OriginalWebSocket(url, protocols);
if (url.includes('engine.kosmi.io') || url.includes('gql-ws')) {
window.__KOSMI_WS_CONNECTED__ = true;
window.__KOSMI_WS__ = socket; // Store the WebSocket reference
const originalAddEventListener = socket.addEventListener.bind(socket);
socket.addEventListener = function(type, listener, options) {
if (type === 'message') {
const wrappedListener = function(event) {
try {
const data = JSON.parse(event.data);
window.__KOSMI_MESSAGE_QUEUE__.push({
timestamp: Date.now(),
data: data
});
} catch (e) {}
return listener.call(this, event);
};
return originalAddEventListener(type, wrappedListener, options);
}
return originalAddEventListener(type, listener, options);
};
let realOnMessage = null;
Object.defineProperty(socket, 'onmessage', {
get: function() { return realOnMessage; },
set: function(handler) {
realOnMessage = function(event) {
try {
const data = JSON.parse(event.data);
window.__KOSMI_MESSAGE_QUEUE__.push({
timestamp: Date.now(),
data: data
});
} catch (e) {}
if (handler) { handler.call(socket, event); }
};
},
configurable: true
});
}
return socket;
};
window.WebSocket.prototype = OriginalWebSocket.prototype;
window.WebSocket.CONNECTING = OriginalWebSocket.CONNECTING;
window.WebSocket.OPEN = OriginalWebSocket.OPEN;
window.WebSocket.CLOSING = OriginalWebSocket.CLOSING;
window.WebSocket.CLOSED = OriginalWebSocket.CLOSED;
window.__KOSMI_WS_HOOK_INSTALLED__ = true;
})();
`
}
// listenForMessages polls for messages from the WebSocket queue
func (c *HybridClient) listenForMessages() {
ticker := time.NewTicker(1 * time.Second)
defer ticker.Stop()
for {
c.mu.RLock()
if !c.connected {
c.mu.RUnlock()
return
}
c.mu.RUnlock()
// Poll for messages
var messages []struct {
Timestamp int64 `json:"timestamp"`
Data map[string]interface{} `json:"data"`
}
script := `
(function() {
if (!window.__KOSMI_MESSAGE_QUEUE__) return [];
const messages = window.__KOSMI_MESSAGE_QUEUE__.slice();
window.__KOSMI_MESSAGE_QUEUE__ = [];
return messages;
})();
`
if err := chromedp.Run(c.ctx, chromedp.Evaluate(script, &messages)); err != nil {
<-ticker.C
continue
}
if len(messages) > 0 {
c.log.Infof("Processing %d messages from queue", len(messages))
}
for _, msg := range messages {
c.processMessage(msg.Data)
}
<-ticker.C
}
}
// processMessage processes a WebSocket message
func (c *HybridClient) processMessage(data map[string]interface{}) {
msgType, ok := data["type"].(string)
if !ok || msgType != "next" {
return
}
payload, ok := data["payload"].(map[string]interface{})
if !ok {
return
}
dataField, ok := payload["data"].(map[string]interface{})
if !ok {
return
}
newMessage, ok := dataField["newMessage"].(map[string]interface{})
if !ok {
return
}
jsonBytes, err := json.Marshal(map[string]interface{}{
"data": map[string]interface{}{
"newMessage": newMessage,
},
})
if err != nil {
return
}
var msgPayload NewMessagePayload
if err := json.Unmarshal(jsonBytes, &msgPayload); err != nil {
return
}
if c.messageCallback != nil {
c.messageCallback(&msgPayload)
}
}
// SendMessage sends a message via WebSocket using browser automation
func (c *HybridClient) SendMessage(text string) error {
c.log.Infof("SendMessage called with text: %s", text)
c.mu.RLock()
if !c.connected {
c.mu.RUnlock()
c.log.Error("SendMessage: not connected")
return fmt.Errorf("not connected")
}
ctx := c.ctx
c.mu.RUnlock()
c.log.Infof("Sending message to room %s via WebSocket", c.roomID)
// Escape the text for JavaScript
escapedText := strings.ReplaceAll(text, `\`, `\\`)
escapedText = strings.ReplaceAll(escapedText, `"`, `\"`)
escapedText = strings.ReplaceAll(escapedText, "\n", `\n`)
// JavaScript to send message via WebSocket
script := fmt.Sprintf(`
(function() {
// Find the Kosmi WebSocket
if (!window.__KOSMI_WS__) {
return { success: false, error: "WebSocket not found" };
}
const ws = window.__KOSMI_WS__;
if (ws.readyState !== WebSocket.OPEN) {
return { success: false, error: "WebSocket not open, state: " + ws.readyState };
}
// GraphQL-WS message format
const message = {
id: "send-" + Date.now(),
type: "start",
payload: {
query: "mutation SendMessage($body: String!, $roomID: ID!) { sendMessage(body: $body, roomID: $roomID) { id body time user { id username displayName } } }",
variables: {
body: "%s",
roomID: "%s"
}
}
};
try {
ws.send(JSON.stringify(message));
return { success: true, message: "Sent via WebSocket" };
} catch (e) {
return { success: false, error: e.toString() };
}
})();
`, escapedText, c.roomID)
var result map[string]interface{}
err := chromedp.Run(ctx,
chromedp.Evaluate(script, &result),
)
if err != nil {
c.log.Errorf("Failed to execute send script: %v", err)
return fmt.Errorf("failed to execute send script: %w", err)
}
c.log.Debugf("Send result: %+v", result)
if success, ok := result["success"].(bool); !ok || !success {
errorMsg := "unknown error"
if errStr, ok := result["error"].(string); ok {
errorMsg = errStr
}
c.log.Errorf("Failed to send message: %s", errorMsg)
return fmt.Errorf("failed to send message: %s", errorMsg)
}
c.log.Infof("✅ Successfully sent message via WebSocket: %s", text)
return nil
}
// OnMessage sets the callback for new messages
func (c *HybridClient) OnMessage(callback func(*NewMessagePayload)) {
c.messageCallback = callback
}
// Disconnect closes the browser and cleans up
func (c *HybridClient) Disconnect() error {
c.mu.Lock()
c.connected = false
c.mu.Unlock()
c.log.Info("Closing hybrid client")
if c.cancel != nil {
c.cancel()
}
return nil
}

11
bridge/kosmi/kosmi.go
View File

@@ -61,8 +61,8 @@ func (b *Bkosmi) Connect() error {
b.roomID = roomID b.roomID = roomID
b.Log.Infof("Extracted room ID: %s", b.roomID) b.Log.Infof("Extracted room ID: %s", b.roomID)
// Create Native client (Playwright establishes WebSocket, we control it directly) // Create GraphQL WebSocket client (pure Go, no Playwright!)
b.client = NewNativeClient(b.roomURL, b.roomID, b.Log) b.client = NewGraphQLWSClient(b.roomURL, b.roomID, b.Log)
// Register message handler // Register message handler
b.client.OnMessage(b.handleIncomingMessage) b.client.OnMessage(b.handleIncomingMessage)
@@ -198,8 +198,11 @@ func extractRoomID(url string) (string, error) {
matches := re.FindStringSubmatch(url) matches := re.FindStringSubmatch(url)
if len(matches) >= 2 { if len(matches) >= 2 {
roomID := matches[1] roomID := matches[1]
// Remove @ prefix if present (Kosmi uses both formats) // Ensure @ prefix is present (required for WebSocket API)
return strings.TrimPrefix(roomID, "@"), nil if !strings.HasPrefix(roomID, "@") {
roomID = "@" + roomID
}
return roomID, nil
} }
} }

521
bridge/kosmi/native_client.go
View File

@@ -1,521 +0,0 @@
package bkosmi
import (
"encoding/json"
"fmt"
"sync"
"time"
"github.com/playwright-community/playwright-go"
"github.com/sirupsen/logrus"
)
// NativeClient uses Playwright to establish WebSocket, then interacts directly via JavaScript
type NativeClient struct {
roomURL string
roomID string
log *logrus.Entry
pw *playwright.Playwright
browser playwright.Browser
page playwright.Page
messageCallback func(*NewMessagePayload)
connected bool
mu sync.RWMutex
}
// NewNativeClient creates a new native client with Playwright-assisted connection
func NewNativeClient(roomURL, roomID string, log *logrus.Entry) *NativeClient {
return &NativeClient{
roomURL: roomURL,
roomID: roomID,
log: log,
}
}
// Connect launches Playwright and establishes the WebSocket connection
func (c *NativeClient) Connect() error {
c.log.Info("Starting Playwright native client")
// Launch Playwright
pw, err := playwright.Run()
if err != nil {
return fmt.Errorf("failed to start Playwright: %w", err)
}
c.pw = pw
// Launch browser with resource optimizations
browser, err := pw.Chromium.Launch(playwright.BrowserTypeLaunchOptions{
Headless: playwright.Bool(true),
Args: []string{
"--no-sandbox",
"--disable-dev-shm-usage",
"--disable-blink-features=AutomationControlled",
// Resource optimizations for reduced CPU/memory usage
"--disable-gpu", // No GPU needed for chat
"--disable-software-rasterizer", // No rendering needed
"--disable-extensions", // No extensions needed
"--disable-background-networking", // No background requests
"--disable-background-timer-throttling",
"--disable-backgrounding-occluded-windows",
"--disable-breakpad", // No crash reporting
"--disable-component-extensions-with-background-pages",
"--disable-features=TranslateUI", // No translation UI
"--disable-ipc-flooding-protection",
"--disable-renderer-backgrounding",
"--force-color-profile=srgb",
"--metrics-recording-only",
"--no-first-run", // Skip first-run tasks
"--mute-audio", // No audio needed
},
})
if err != nil {
c.pw.Stop()
return fmt.Errorf("failed to launch browser: %w", err)
}
c.browser = browser
// Create context
context, err := browser.NewContext(playwright.BrowserNewContextOptions{
UserAgent: playwright.String("Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"),
})
if err != nil {
c.browser.Close()
c.pw.Stop()
return fmt.Errorf("failed to create context: %w", err)
}
// Create page
page, err := context.NewPage()
if err != nil {
c.browser.Close()
c.pw.Stop()
return fmt.Errorf("failed to create page: %w", err)
}
c.page = page
// Inject WebSocket interceptor
c.log.Debug("Injecting WebSocket access layer")
if err := c.injectWebSocketAccess(); err != nil {
c.Disconnect()
return fmt.Errorf("failed to inject WebSocket access: %w", err)
}
// Navigate to room
c.log.Infof("Navigating to Kosmi room: %s", c.roomURL)
if _, err := page.Goto(c.roomURL, playwright.PageGotoOptions{
WaitUntil: playwright.WaitUntilStateDomcontentloaded, // Wait for DOM only, not all resources
}); err != nil {
c.Disconnect()
return fmt.Errorf("failed to navigate: %w", err)
}
// Wait for WebSocket to establish
c.log.Debug("Waiting for WebSocket connection")
if err := c.waitForWebSocket(); err != nil {
c.Disconnect()
return fmt.Errorf("WebSocket not established: %w", err)
}
// Subscribe to room messages
c.log.Debugf("Subscribing to messages in room %s", c.roomID)
if err := c.subscribeToMessages(); err != nil {
c.Disconnect()
return fmt.Errorf("failed to subscribe: %w", err)
}
c.mu.Lock()
c.connected = true
c.mu.Unlock()
c.log.Info("Native client connected successfully")
// Start message listener
go c.listenForMessages()
return nil
}
// injectWebSocketAccess injects JavaScript that provides direct WebSocket access
func (c *NativeClient) injectWebSocketAccess() error {
script := `
(function() {
if (window.__KOSMI_NATIVE_CLIENT__) return;
const OriginalWebSocket = window.WebSocket;
window.__KOSMI_WS__ = null;
window.__KOSMI_MESSAGE_QUEUE__ = [];
window.__KOSMI_READY__ = false;
// Hook WebSocket constructor to capture the connection
window.WebSocket = function(url, protocols) {
const socket = new OriginalWebSocket(url, protocols);
if (url.includes('engine.kosmi.io') || url.includes('gql-ws')) {
window.__KOSMI_WS__ = socket;
// Hook message handler to queue messages
socket.addEventListener('message', (event) => {
try {
const data = JSON.parse(event.data);
window.__KOSMI_MESSAGE_QUEUE__.push({
timestamp: Date.now(),
data: data
});
} catch (e) {
// Ignore non-JSON messages
}
});
// Mark as ready when connection opens
socket.addEventListener('open', () => {
window.__KOSMI_READY__ = true;
});
}
return socket;
};
// Preserve WebSocket properties
window.WebSocket.prototype = OriginalWebSocket.prototype;
window.WebSocket.CONNECTING = OriginalWebSocket.CONNECTING;
window.WebSocket.OPEN = OriginalWebSocket.OPEN;
window.WebSocket.CLOSING = OriginalWebSocket.CLOSING;
window.WebSocket.CLOSED = OriginalWebSocket.CLOSED;
window.__KOSMI_NATIVE_CLIENT__ = true;
})();
`
return c.page.AddInitScript(playwright.Script{
Content: playwright.String(script),
})
}
// waitForWebSocket waits for the WebSocket to be established
func (c *NativeClient) waitForWebSocket() error {
for i := 0; i < 30; i++ { // 15 seconds max
result, err := c.page.Evaluate(`
(function() {
return {
ready: !!window.__KOSMI_READY__,
wsExists: !!window.__KOSMI_WS__,
wsState: window.__KOSMI_WS__ ? window.__KOSMI_WS__.readyState : -1
};
})();
`)
if err != nil {
return err
}
status := result.(map[string]interface{})
ready := status["ready"].(bool)
if ready {
c.log.Info("✅ WebSocket is ready")
return nil
}
if i%5 == 0 {
c.log.Debugf("Waiting for WebSocket... (attempt %d/30)", i+1)
}
time.Sleep(500 * time.Millisecond)
}
return fmt.Errorf("timeout waiting for WebSocket")
}
// subscribeToMessages subscribes to room messages via the WebSocket
func (c *NativeClient) subscribeToMessages() error {
script := fmt.Sprintf(`
(function() {
if (!window.__KOSMI_WS__ || window.__KOSMI_WS__.readyState !== WebSocket.OPEN) {
return { success: false, error: 'WebSocket not ready' };
}
const subscription = {
id: 'native-client-subscription',
type: 'subscribe',
payload: {
query: 'subscription { newMessage(roomId: "%s") { body time user { displayName username } } }',
variables: {}
}
};
try {
window.__KOSMI_WS__.send(JSON.stringify(subscription));
return { success: true };
} catch (e) {
return { success: false, error: e.toString() };
}
})();
`, c.roomID)
result, err := c.page.Evaluate(script)
if err != nil {
return err
}
response := result.(map[string]interface{})
if success, ok := response["success"].(bool); !ok || !success {
errMsg := "unknown error"
if e, ok := response["error"].(string); ok {
errMsg = e
}
return fmt.Errorf("subscription failed: %s", errMsg)
}
return nil
}
// listenForMessages continuously polls for new messages
func (c *NativeClient) listenForMessages() {
ticker := time.NewTicker(500 * time.Millisecond)
defer ticker.Stop()
for {
c.mu.RLock()
if !c.connected {
c.mu.RUnlock()
return
}
c.mu.RUnlock()
if err := c.pollMessages(); err != nil {
c.log.Errorf("Error polling messages: %v", err)
}
<-ticker.C
}
}
// pollMessages retrieves and processes messages from the queue
func (c *NativeClient) pollMessages() error {
result, err := c.page.Evaluate(`
(function() {
if (!window.__KOSMI_MESSAGE_QUEUE__) return null;
if (window.__KOSMI_MESSAGE_QUEUE__.length === 0) return null;
const messages = window.__KOSMI_MESSAGE_QUEUE__.slice();
window.__KOSMI_MESSAGE_QUEUE__ = [];
return messages;
})();
`)
if err != nil {
return err
}
// Early return if no messages (reduces CPU during idle)
if result == nil {
return nil
}
messagesJSON, err := json.Marshal(result)
if err != nil {
return err
}
var messages []struct {
Timestamp int64 `json:"timestamp"`
Data map[string]interface{} `json:"data"`
}
if err := json.Unmarshal(messagesJSON, &messages); err != nil {
return err
}
for _, msg := range messages {
c.processMessage(msg.Data)
}
return nil
}
// processMessage processes a single WebSocket message
func (c *NativeClient) processMessage(data map[string]interface{}) {
msgType, ok := data["type"].(string)
if !ok || msgType != "next" {
return
}
payload, ok := data["payload"].(map[string]interface{})
if !ok {
return
}
dataField, ok := payload["data"].(map[string]interface{})
if !ok {
return
}
newMessage, ok := dataField["newMessage"].(map[string]interface{})
if !ok {
return
}
// Parse into our struct
jsonBytes, err := json.Marshal(map[string]interface{}{
"data": map[string]interface{}{
"newMessage": newMessage,
},
})
if err != nil {
return
}
var msgPayload NewMessagePayload
if err := json.Unmarshal(jsonBytes, &msgPayload); err != nil {
return
}
if c.messageCallback != nil {
c.messageCallback(&msgPayload)
}
}
// SendMessage sends a message by typing into the Kosmi chat input field
func (c *NativeClient) SendMessage(text string) error {
c.mu.RLock()
if !c.connected {
c.mu.RUnlock()
return fmt.Errorf("not connected")
}
c.mu.RUnlock()
c.log.Debugf("Sending message to Kosmi: %s", text)
// Escape the message text for JavaScript
textJSON, _ := json.Marshal(text)
script := fmt.Sprintf(`
(async function() {
try {
// Try multiple strategies to find the chat input
let input = null;
// Strategy 1: Look for textarea
const textareas = document.querySelectorAll('textarea');
for (let ta of textareas) {
if (ta.offsetParent !== null) { // visible
input = ta;
break;
}
}
// Strategy 2: Look for contenteditable
if (!input) {
const editables = document.querySelectorAll('[contenteditable="true"]');
for (let ed of editables) {
if (ed.offsetParent !== null) { // visible
input = ed;
break;
}
}
}
// Strategy 3: Look for input text
if (!input) {
const inputs = document.querySelectorAll('input[type="text"]');
for (let inp of inputs) {
if (inp.offsetParent !== null) { // visible
input = inp;
break;
}
}
}
if (!input) {
return { success: false, error: 'Could not find any visible input element' };
}
// Set the value based on element type
if (input.contentEditable === 'true') {
input.textContent = %s;
input.dispatchEvent(new Event('input', { bubbles: true }));
} else {
input.value = %s;
input.dispatchEvent(new Event('input', { bubbles: true }));
input.dispatchEvent(new Event('change', { bubbles: true }));
}
// Focus the input
input.focus();
// Wait a tiny bit
await new Promise(resolve => setTimeout(resolve, 100));
// Find and click the send button, or press Enter
const sendButton = document.querySelector('button[type="submit"], button[class*="send" i], button[aria-label*="send" i]');
if (sendButton && sendButton.offsetParent !== null) {
sendButton.click();
} else {
// Simulate Enter key press
const enterEvent = new KeyboardEvent('keydown', {
key: 'Enter',
code: 'Enter',
keyCode: 13,
which: 13,
bubbles: true,
cancelable: true
});
input.dispatchEvent(enterEvent);
}
return { success: true };
} catch (e) {
return { success: false, error: e.toString() };
}
})();
`, string(textJSON), string(textJSON))
result, err := c.page.Evaluate(script)
if err != nil {
c.log.Errorf("Failed to execute send script: %v", err)
return fmt.Errorf("failed to execute send: %w", err)
}
response := result.(map[string]interface{})
if success, ok := response["success"].(bool); !ok || !success {
errMsg := "unknown error"
if e, ok := response["error"].(string); ok {
errMsg = e
}
c.log.Errorf("Send failed: %s", errMsg)
return fmt.Errorf("send failed: %s", errMsg)
}
c.log.Debug("Successfully sent message to Kosmi")
return nil
}
// OnMessage registers a callback for incoming messages
func (c *NativeClient) OnMessage(callback func(*NewMessagePayload)) {
c.messageCallback = callback
}
// Disconnect closes the Playwright browser
func (c *NativeClient) Disconnect() error {
c.mu.Lock()
c.connected = false
c.mu.Unlock()
c.log.Debug("Closing Playwright browser")
if c.browser != nil {
c.browser.Close()
}
if c.pw != nil {
c.pw.Stop()
}
return nil
}
// IsConnected returns whether the client is connected
func (c *NativeClient) IsConnected() bool {
c.mu.RLock()
defer c.mu.RUnlock()
return c.connected
}

347
bridge/kosmi/playwright_client.go
View File

@@ -1,347 +0,0 @@
package bkosmi
import (
"encoding/json"
"fmt"
"sync"
"time"
"github.com/playwright-community/playwright-go"
"github.com/sirupsen/logrus"
)
// PlaywrightClient manages a Playwright browser instance to connect to Kosmi
type PlaywrightClient struct {
roomURL string
log *logrus.Entry
pw *playwright.Playwright
browser playwright.Browser
page playwright.Page
messageCallback func(*NewMessagePayload)
connected bool
mu sync.RWMutex
}
// NewPlaywrightClient creates a new Playwright-based Kosmi client
func NewPlaywrightClient(roomURL string, log *logrus.Entry) *PlaywrightClient {
return &PlaywrightClient{
roomURL: roomURL,
log: log,
}
}
// Connect launches Playwright and navigates to the Kosmi room
func (c *PlaywrightClient) Connect() error {
c.log.Info("Launching Playwright browser for Kosmi connection")
// Create Playwright instance (using system Chromium, no install needed)
pw, err := playwright.Run()
if err != nil {
return fmt.Errorf("failed to start Playwright: %w", err)
}
c.pw = pw
// Launch browser using system Chromium
browser, err := pw.Chromium.Launch(playwright.BrowserTypeLaunchOptions{
Headless: playwright.Bool(true),
ExecutablePath: playwright.String("/usr/bin/chromium"),
Args: []string{
"--no-sandbox",
"--disable-dev-shm-usage",
"--disable-blink-features=AutomationControlled",
},
})
if err != nil {
return fmt.Errorf("failed to launch browser: %w", err)
}
c.browser = browser
// Create context and page
context, err := browser.NewContext(playwright.BrowserNewContextOptions{
UserAgent: playwright.String("Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"),
})
if err != nil {
return fmt.Errorf("failed to create context: %w", err)
}
page, err := context.NewPage()
if err != nil {
return fmt.Errorf("failed to create page: %w", err)
}
c.page = page
// Inject WebSocket hook before navigation
c.log.Info("Injecting WebSocket interceptor...")
if err := c.injectWebSocketHook(); err != nil {
return fmt.Errorf("failed to inject WebSocket hook: %w", err)
}
// Navigate to the room
c.log.Infof("Navigating to Kosmi room: %s", c.roomURL)
if _, err := page.Goto(c.roomURL, playwright.PageGotoOptions{
WaitUntil: playwright.WaitUntilStateNetworkidle,
}); err != nil {
return fmt.Errorf("failed to navigate to room: %w", err)
}
// Wait for page to be ready
if err := page.WaitForLoadState(playwright.PageWaitForLoadStateOptions{
State: playwright.LoadStateNetworkidle,
}); err != nil {
c.log.Warnf("Page load state warning: %v", err)
}
c.log.Info("Page loaded, waiting for WebSocket connection...")
time.Sleep(3 * time.Second)
c.mu.Lock()
c.connected = true
c.mu.Unlock()
c.log.Info("Successfully connected to Kosmi via Playwright")
// Start message listener
go c.listenForMessages()
return nil
}
// injectWebSocketHook injects the WebSocket interception script
func (c *PlaywrightClient) injectWebSocketHook() error {
script := c.getWebSocketHookScript()
return c.page.AddInitScript(playwright.Script{
Content: playwright.String(script),
})
}
// getWebSocketHookScript returns the JavaScript to hook WebSocket
func (c *PlaywrightClient) getWebSocketHookScript() string {
return `
(function() {
if (window.__KOSMI_WS_HOOK_INSTALLED__) {
return;
}
const OriginalWebSocket = window.WebSocket;
window.__KOSMI_MESSAGE_QUEUE__ = [];
window.WebSocket = function(url, protocols) {
const socket = new OriginalWebSocket(url, protocols);
if (url.includes('engine.kosmi.io') || url.includes('gql-ws')) {
console.log('[Kosmi Bridge] WebSocket hook active for:', url);
window.__KOSMI_WS_CONNECTED__ = true;
const originalAddEventListener = socket.addEventListener.bind(socket);
socket.addEventListener = function(type, listener, options) {
if (type === 'message') {
const wrappedListener = function(event) {
try {
const data = JSON.parse(event.data);
window.__KOSMI_MESSAGE_QUEUE__.push({
timestamp: Date.now(),
data: data
});
} catch (e) {}
return listener.call(this, event);
};
return originalAddEventListener(type, wrappedListener, options);
}
return originalAddEventListener(type, listener, options);
};
let realOnMessage = null;
Object.defineProperty(socket, 'onmessage', {
get: function() { return realOnMessage; },
set: function(handler) {
realOnMessage = function(event) {
try {
const data = JSON.parse(event.data);
window.__KOSMI_MESSAGE_QUEUE__.push({
timestamp: Date.now(),
data: data
});
} catch (e) {}
if (handler) { handler.call(socket, event); }
};
},
configurable: true
});
}
return socket;
};
window.WebSocket.prototype = OriginalWebSocket.prototype;
window.WebSocket.CONNECTING = OriginalWebSocket.CONNECTING;
window.WebSocket.OPEN = OriginalWebSocket.OPEN;
window.WebSocket.CLOSING = OriginalWebSocket.CLOSING;
window.WebSocket.CLOSED = OriginalWebSocket.CLOSED;
window.__KOSMI_WS_HOOK_INSTALLED__ = true;
console.log('[Kosmi Bridge] WebSocket hook installed');
})();
`
}
// listenForMessages polls for new messages from the WebSocket queue
func (c *PlaywrightClient) listenForMessages() {
ticker := time.NewTicker(1 * time.Second)
defer ticker.Stop()
for {
c.mu.RLock()
if !c.connected {
c.mu.RUnlock()
return
}
c.mu.RUnlock()
// Poll for messages
result, err := c.page.Evaluate(`
(function() {
if (!window.__KOSMI_MESSAGE_QUEUE__) return [];
const messages = window.__KOSMI_MESSAGE_QUEUE__.slice();
window.__KOSMI_MESSAGE_QUEUE__ = [];
return messages;
})();
`)
if err != nil {
c.log.Debugf("Error polling messages: %v", err)
<-ticker.C
continue
}
// Parse messages
var messages []struct {
Timestamp int64 `json:"timestamp"`
Data map[string]interface{} `json:"data"`
}
if err := json.Unmarshal([]byte(fmt.Sprintf("%v", result)), &messages); err != nil {
<-ticker.C
continue
}
if len(messages) > 0 {
c.log.Infof("Processing %d messages from queue", len(messages))
}
for _, msg := range messages {
c.processMessage(msg.Data)
}
<-ticker.C
}
}
// processMessage processes a single WebSocket message
func (c *PlaywrightClient) processMessage(data map[string]interface{}) {
// Check if this is a newMessage subscription event
msgType, ok := data["type"].(string)
if !ok || msgType != "next" {
return
}
payload, ok := data["payload"].(map[string]interface{})
if !ok {
return
}
dataField, ok := payload["data"].(map[string]interface{})
if !ok {
return
}
newMessage, ok := dataField["newMessage"].(map[string]interface{})
if !ok {
return
}
// Parse into our struct
jsonBytes, err := json.Marshal(map[string]interface{}{
"data": map[string]interface{}{
"newMessage": newMessage,
},
})
if err != nil {
return
}
var msgPayload NewMessagePayload
if err := json.Unmarshal(jsonBytes, &msgPayload); err != nil {
return
}
// Call the callback
if c.messageCallback != nil {
c.messageCallback(&msgPayload)
}
}
// SendMessage sends a message to the Kosmi chat
func (c *PlaywrightClient) SendMessage(text string) error {
c.mu.RLock()
if !c.connected {
c.mu.RUnlock()
return fmt.Errorf("not connected")
}
c.mu.RUnlock()
selector := `div[role="textbox"][contenteditable="true"]`
// Wait for the input to be available
_, err := c.page.WaitForSelector(selector, playwright.PageWaitForSelectorOptions{
Timeout: playwright.Float(5000),
})
if err != nil {
return fmt.Errorf("chat input not available: %w", err)
}
// Get the input element
input := c.page.Locator(selector)
// Clear and type the message
if err := input.Fill(text); err != nil {
return fmt.Errorf("failed to fill message: %w", err)
}
// Press Enter to send
if err := input.Press("Enter"); err != nil {
return fmt.Errorf("failed to press Enter: %w", err)
}
c.log.Debugf("Sent message: %s", text)
return nil
}
// OnMessage sets the callback for new messages
func (c *PlaywrightClient) OnMessage(callback func(*NewMessagePayload)) {
c.messageCallback = callback
}
// Disconnect closes the browser
func (c *PlaywrightClient) Disconnect() error {
c.mu.Lock()
c.connected = false
c.mu.Unlock()
c.log.Info("Closing Playwright browser")
if c.browser != nil {
if err := c.browser.Close(); err != nil {
c.log.Warnf("Error closing browser: %v", err)
}
}
if c.pw != nil {
if err := c.pw.Stop(); err != nil {
c.log.Warnf("Error stopping Playwright: %v", err)
}
}
return nil
}

12
docker-compose.yml
View File

@@ -16,17 +16,11 @@ services:
# ports: # ports:
# - "4242:4242" # - "4242:4242"
environment: environment:
# Chrome/Chromium configuration for headless mode
- CHROME_BIN=/usr/bin/chromium
- CHROME_PATH=/usr/bin/chromium
# Optional: Set timezone # Optional: Set timezone
- TZ=America/New_York - TZ=America/New_York
# Security options for Chrome in Docker # Optional: Set memory limits (much lower now without browser!)
security_opt: # mem_limit: 128m
- seccomp:unconfined # mem_reservation: 64m
# Optional: Set memory limits
# mem_limit: 512m
# mem_reservation: 256m
logging: logging:
driver: "json-file" driver: "json-file"
options: options: