feat: add Google Chat channel with Pub/Sub pull, Cards V2, Drive upload

cmd/gateway.go

@@ -18,6 +18,7 @@ import (
 	"github.com/nextlevelbuilder/goclaw/internal/channels"
 	"github.com/nextlevelbuilder/goclaw/internal/channels/discord"
 	"github.com/nextlevelbuilder/goclaw/internal/channels/feishu"
+	"github.com/nextlevelbuilder/goclaw/internal/channels/googlechat"
 	slackchannel "github.com/nextlevelbuilder/goclaw/internal/channels/slack"
 	"github.com/nextlevelbuilder/goclaw/internal/channels/telegram"
 	"github.com/nextlevelbuilder/goclaw/internal/channels/whatsapp"
@@ -689,7 +690,7 @@ func runGateway() {
 	if mcpMgr != nil {
 		mcpToolLister = mcpMgr
 	}
-	agentsH, skillsH, tracesH, mcpH, customToolsH, channelInstancesH, providersH, delegationsH, builtinToolsH, pendingMessagesH := wireHTTP(pgStores, cfg.Gateway.Token, msgBus, toolsReg, providerRegistry, permPE.IsOwner, gatewayAddr, mcpToolLister)
+	agentsH, skillsH, tracesH, mcpH, customToolsH, channelInstancesH, providersH, delegationsH, builtinToolsH, pendingMessagesH, projectsH := wireHTTP(pgStores, cfg.Gateway.Token, msgBus, toolsReg, providerRegistry, permPE.IsOwner, gatewayAddr, mcpToolLister)
 	if agentsH != nil {
 		server.SetAgentsHandler(agentsH)
 	}
@@ -705,6 +706,9 @@ func runGateway() {
 	if mcpH != nil {
 		server.SetMCPHandler(mcpH)
 	}
+	if projectsH != nil {
+		server.SetProjectHandler(projectsH)
+	}
 	if customToolsH != nil {
 		server.SetCustomToolsHandler(customToolsH)
 	}
@@ -810,6 +814,7 @@ func runGateway() {
 		instanceLoader.RegisterFactory(channels.TypeZaloPersonal, zalopersonal.FactoryWithPendingStore(pgStores.PendingMessages))
 		instanceLoader.RegisterFactory(channels.TypeWhatsApp, whatsapp.Factory)
 		instanceLoader.RegisterFactory(channels.TypeSlack, slackchannel.FactoryWithPendingStore(pgStores.PendingMessages))
+		instanceLoader.RegisterFactory(channels.TypeGoogleChat, googlechat.FactoryWithPendingStore(pgStores.PendingMessages))
 		if err := instanceLoader.LoadAll(context.Background()); err != nil {
 			slog.Error("failed to load channel instances from DB", "error", err)
 		}

cmd/gateway_channels_setup.go

@@ -13,6 +13,7 @@ import (
 	"github.com/nextlevelbuilder/goclaw/internal/channels"
 	"github.com/nextlevelbuilder/goclaw/internal/channels/discord"
 	"github.com/nextlevelbuilder/goclaw/internal/channels/feishu"
+	"github.com/nextlevelbuilder/goclaw/internal/channels/googlechat"
 	slackchannel "github.com/nextlevelbuilder/goclaw/internal/channels/slack"
 	"github.com/nextlevelbuilder/goclaw/internal/channels/telegram"
 	"github.com/nextlevelbuilder/goclaw/internal/channels/whatsapp"
@@ -97,6 +98,16 @@ func registerConfigChannels(cfg *config.Config, channelMgr *channels.Manager, ms
 			slog.Info("feishu/lark channel enabled (config)")
 		}
 	}
+
+	if cfg.Channels.GoogleChat.Enabled && cfg.Channels.GoogleChat.ServiceAccountFile != "" && instanceLoader == nil {
+		gc, err := googlechat.New(cfg.Channels.GoogleChat, msgBus, nil)
+		if err != nil {
+			slog.Error("failed to initialize google chat channel", "error", err)
+		} else {
+			channelMgr.RegisterChannel(channels.TypeGoogleChat, gc)
+			slog.Info("google chat channel enabled (config)")
+		}
+	}
 }
 
 // wireChannelRPCMethods registers WS RPC methods for channels, instances, agent links, and teams.

internal/channels/channel.go

@@ -58,6 +58,7 @@ const (
 	TypeWhatsApp     = "whatsapp"
 	TypeZaloOA       = "zalo_oa"
 	TypeZaloPersonal = "zalo_personal"
+	TypeGoogleChat   = "google_chat"
 )
 
 // Channel defines the interface that all channel implementations must satisfy.

internal/channels/googlechat/auth.go

@@ -0,0 +1,170 @@
+package googlechat
+
+import (
+	"context"
+	"crypto"
+	"crypto/rand"
+	"crypto/rsa"
+	"crypto/sha256"
+	"crypto/x509"
+	"encoding/base64"
+	"encoding/json"
+	"encoding/pem"
+	"fmt"
+	"io"
+	"net/http"
+	"net/url"
+	"os"
+	"strings"
+	"sync"
+	"time"
+)
+
+type ServiceAccountAuth struct {
+	email         string
+	privateKey    *rsa.PrivateKey
+	scopes        []string
+	token         string
+	expiresAt     time.Time
+	mu            sync.Mutex
+	tokenEndpoint string
+	httpClient    *http.Client
+}
+
+type serviceAccountFile struct {
+	Type        string `json:"type"`
+	ClientEmail string `json:"client_email"`
+	PrivateKey  string `json:"private_key"`
+	TokenURI    string `json:"token_uri"`
+}
+
+func NewServiceAccountAuth(saFilePath string, scopes []string) (*ServiceAccountAuth, error) {
+	data, err := os.ReadFile(saFilePath)
+	if err != nil {
+		return nil, fmt.Errorf("read service account file: %w", err)
+	}
+
+	var sa serviceAccountFile
+	if err := json.Unmarshal(data, &sa); err != nil {
+		return nil, fmt.Errorf("parse service account file: %w", err)
+	}
+	if sa.ClientEmail == "" {
+		return nil, fmt.Errorf("service account file missing client_email")
+	}
+	if sa.PrivateKey == "" {
+		return nil, fmt.Errorf("service account file missing private_key")
+	}
+
+	block, _ := pem.Decode([]byte(sa.PrivateKey))
+	if block == nil {
+		return nil, fmt.Errorf("failed to decode PEM block from private_key")
+	}
+
+	key, err := x509.ParsePKCS8PrivateKey(block.Bytes)
+	if err != nil {
+		rsaKey, err2 := x509.ParsePKCS1PrivateKey(block.Bytes)
+		if err2 != nil {
+			return nil, fmt.Errorf("parse private key: %w (pkcs1: %w)", err, err2)
+		}
+		key = rsaKey
+	}
+
+	rsaKey, ok := key.(*rsa.PrivateKey)
+	if !ok {
+		return nil, fmt.Errorf("private key is not RSA")
+	}
+
+	ep := sa.TokenURI
+	if ep == "" {
+		ep = tokenEndpoint
+	}
+
+	return &ServiceAccountAuth{
+		email:         sa.ClientEmail,
+		privateKey:    rsaKey,
+		scopes:        scopes,
+		tokenEndpoint: ep,
+		httpClient:    &http.Client{Timeout: 10 * time.Second},
+	}, nil
+}
+
+func (a *ServiceAccountAuth) Token(ctx context.Context) (string, error) {
+	a.mu.Lock()
+	defer a.mu.Unlock()
+
+	if a.token != "" && time.Now().Add(60*time.Second).Before(a.expiresAt) {
+		return a.token, nil
+	}
+
+	now := time.Now()
+	claims := map[string]any{
+		"iss":   a.email,
+		"scope": strings.Join(a.scopes, " "),
+		"aud":   tokenEndpoint,
+		"iat":   now.Unix(),
+		"exp":   now.Add(time.Hour).Unix(),
+	}
+
+	signedJWT, err := signJWT(a.privateKey, claims)
+	if err != nil {
+		return "", fmt.Errorf("sign JWT: %w", err)
+	}
+
+	form := url.Values{
+		"grant_type": {"urn:ietf:params:oauth:grant-type:jwt-bearer"},
+		"assertion":  {signedJWT},
+	}
+
+	req, err := http.NewRequestWithContext(ctx, "POST", a.tokenEndpoint, strings.NewReader(form.Encode()))
+	if err != nil {
+		return "", err
+	}
+	req.Header.Set("Content-Type", "application/x-www-form-urlencoded")
+
+	resp, err := a.httpClient.Do(req)
+	if err != nil {
+		return "", fmt.Errorf("token exchange request: %w", err)
+	}
+	defer resp.Body.Close()
+
+	body, _ := io.ReadAll(resp.Body)
+	if resp.StatusCode != http.StatusOK {
+		return "", fmt.Errorf("token exchange failed (%d): %s", resp.StatusCode, string(body))
+	}
+
+	var tokenResp struct {
+		AccessToken string `json:"access_token"`
+		ExpiresIn   int    `json:"expires_in"`
+		TokenType   string `json:"token_type"`
+	}
+	if err := json.Unmarshal(body, &tokenResp); err != nil {
+		return "", fmt.Errorf("parse token response: %w", err)
+	}
+
+	a.token = tokenResp.AccessToken
+	a.expiresAt = now.Add(time.Duration(tokenResp.ExpiresIn) * time.Second)
+
+	return a.token, nil
+}
+
+func signJWT(key *rsa.PrivateKey, claims map[string]any) (string, error) {
+	header := base64URLEncode([]byte(`{"alg":"RS256","typ":"JWT"}`))
+	payload, err := json.Marshal(claims)
+	if err != nil {
+		return "", err
+	}
+	payloadEnc := base64URLEncode(payload)
+	signingInput := header + "." + payloadEnc
+
+	hash := sha256.Sum256([]byte(signingInput))
+	sig, err := rsa.SignPKCS1v15(rand.Reader, key, crypto.SHA256, hash[:])
+	if err != nil {
+		return "", err
+	}
+
+	return signingInput + "." + base64URLEncode(sig), nil
+}
+
+func base64URLEncode(data []byte) string {
+	return base64.RawURLEncoding.EncodeToString(data)
+}

internal/channels/googlechat/auth_test.go

@@ -0,0 +1,145 @@
+package googlechat
+
+import (
+	"context"
+	"crypto/rand"
+	"crypto/rsa"
+	"crypto/x509"
+	"encoding/json"
+	"encoding/pem"
+	"net/http"
+	"net/http/httptest"
+	"os"
+	"path/filepath"
+	"testing"
+	"time"
+)
+
+func testServiceAccountJSON(t *testing.T, dir string) (string, *rsa.PrivateKey) {
+	t.Helper()
+	key, err := rsa.GenerateKey(rand.Reader, 2048)
+	if err != nil {
+		t.Fatal(err)
+	}
+	pkcs8, err := x509.MarshalPKCS8PrivateKey(key)
+	if err != nil {
+		t.Fatal(err)
+	}
+	pemBlock := pem.EncodeToMemory(&pem.Block{Type: "PRIVATE KEY", Bytes: pkcs8})
+
+	sa := map[string]string{
+		"type":         "service_account",
+		"client_email": "test@test.iam.gserviceaccount.com",
+		"private_key":  string(pemBlock),
+		"token_uri":    "https://oauth2.googleapis.com/token",
+	}
+	data, _ := json.Marshal(sa)
+	path := filepath.Join(dir, "sa.json")
+	if err := os.WriteFile(path, data, 0600); err != nil {
+		t.Fatal(err)
+	}
+	return path, key
+}
+
+func TestNewServiceAccountAuth_ValidFile(t *testing.T) {
+	dir := t.TempDir()
+	path, _ := testServiceAccountJSON(t, dir)
+	auth, err := NewServiceAccountAuth(path, []string{scopeChat})
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if auth.email != "test@test.iam.gserviceaccount.com" {
+		t.Errorf("email = %q, want test@test.iam.gserviceaccount.com", auth.email)
+	}
+}
+
+func TestNewServiceAccountAuth_InvalidFile(t *testing.T) {
+	dir := t.TempDir()
+	path := filepath.Join(dir, "bad.json")
+	os.WriteFile(path, []byte("{bad json"), 0600)
+	_, err := NewServiceAccountAuth(path, []string{scopeChat})
+	if err == nil {
+		t.Fatal("expected error for invalid JSON")
+	}
+}
+
+func TestNewServiceAccountAuth_MissingFile(t *testing.T) {
+	_, err := NewServiceAccountAuth("/nonexistent/sa.json", []string{scopeChat})
+	if err == nil {
+		t.Fatal("expected error for missing file")
+	}
+}
+
+func TestServiceAccountAuth_Token_CachesWithinTTL(t *testing.T) {
+	dir := t.TempDir()
+	path, _ := testServiceAccountJSON(t, dir)
+	callCount := 0
+	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		callCount++
+		json.NewEncoder(w).Encode(map[string]any{
+			"access_token": "tok-123",
+			"expires_in":   3600,
+			"token_type":   "Bearer",
+		})
+	}))
+	defer ts.Close()
+
+	auth, err := NewServiceAccountAuth(path, []string{scopeChat})
+	if err != nil {
+		t.Fatal(err)
+	}
+	auth.tokenEndpoint = ts.URL
+
+	ctx := context.Background()
+	tok1, _ := auth.Token(ctx)
+	tok2, _ := auth.Token(ctx)
+	if tok1 != tok2 {
+		t.Errorf("tokens differ")
+	}
+	if callCount != 1 {
+		t.Errorf("callCount = %d, want 1", callCount)
+	}
+}
+
+func TestServiceAccountAuth_Token_RefreshesExpired(t *testing.T) {
+	dir := t.TempDir()
+	path, _ := testServiceAccountJSON(t, dir)
+	callCount := 0
+	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		callCount++
+		json.NewEncoder(w).Encode(map[string]any{
+			"access_token": "tok",
+			"expires_in":   1,
+			"token_type":   "Bearer",
+		})
+	}))
+	defer ts.Close()
+
+	auth, err := NewServiceAccountAuth(path, []string{scopeChat})
+	if err != nil {
+		t.Fatal(err)
+	}
+	auth.tokenEndpoint = ts.URL
+	auth.Token(context.Background())
+	auth.expiresAt = time.Now().Add(-1 * time.Minute)
+	auth.Token(context.Background())
+	if callCount != 2 {
+		t.Errorf("callCount = %d, want 2", callCount)
+	}
+}
+
+func TestServiceAccountAuth_Token_RefreshFailure(t *testing.T) {
+	dir := t.TempDir()
+	path, _ := testServiceAccountJSON(t, dir)
+	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.WriteHeader(500)
+	}))
+	defer ts.Close()
+
+	auth, _ := NewServiceAccountAuth(path, []string{scopeChat})
+	auth.tokenEndpoint = ts.URL
+	_, err := auth.Token(context.Background())
+	if err == nil {
+		t.Fatal("expected error on 500")
+	}
+}