feat: forwarded message handler for the login code

Makefile

@@ -10,6 +10,7 @@ proto:
 	go install google.golang.org/grpc/cmd/protoc-gen-go-grpc@v1.5.1
 	PATH=${PATH}:~/go/bin protoc --go_out=plugins=grpc:. --go_opt=paths=source_relative \
 		api/grpc/admin/*.proto \
+		api/grpc/source-telegram/*.proto \
 		api/grpc/tgbot/*.proto \
 		api/grpc/messages/*.proto \
 		api/grpc/queue/*.proto

README.md

@@ -5,5 +5,5 @@ kubectl create secret generic bot-telegram \
   --from-literal=telegram=<TGBOT_TOKEN> \
   --from-literal=payment=<PAYMENT_TOKEN> \
   --from-literal=donation=<DONATION_CHAT_ID> \
-  --from-literal=loginCodeFromUserIds=<USER_ID_1>:true,<USER_ID_2>:true,...
+  --from-literal=loginCodeFromUserIds=<USER_ID_1>:<REPLICA_NUM_1>,<USER_ID_2>:<REPLICA_NUM_2>,...
 ```

api/grpc/source-telegram/service.go

@@ -0,0 +1,55 @@
+package source_telegram
+
+import (
+	"context"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
+)
+
+type Service interface {
+	Login(ctx context.Context, code int64, replicaIdx uint) (err error)
+}
+
+type service struct {
+	uri string
+}
+
+func NewService(uri string) Service {
+	return service{
+		uri: uri,
+	}
+}
+
+func (svc service) Login(ctx context.Context, code int64, replicaIdx uint) (err error) {
+	req := &LoginRequest{
+		Code:         uint32(code),
+		ReplicaIndex: uint32(replicaIdx),
+	}
+	creds := grpc.WithTransportCredentials(insecure.NewCredentials())
+	var success bool
+	for {
+		success, err = svc.loginOnce(ctx, req, creds)
+		if err != nil {
+			break
+		}
+		if success {
+			break
+		}
+	}
+	return
+}
+
+func (svc service) loginOnce(ctx context.Context, req *LoginRequest, creds grpc.DialOption) (success bool, err error) {
+	var conn *grpc.ClientConn
+	conn, err = grpc.NewClient(svc.uri, creds)
+	var resp *LoginResponse
+	if err == nil {
+		defer conn.Close()
+		client := NewServiceClient(conn)
+		resp, err = client.Login(ctx, req)
+	}
+	if err == nil {
+		success = resp.Success
+	}
+	return
+}

api/grpc/source-telegram/service.proto

@@ -0,0 +1,18 @@
+syntax = "proto3";
+
+package awakari.source.telegram;
+
+option go_package = "./api/grpc/source-telegram";
+
+service Service {
+  rpc Login(LoginRequest) returns (LoginResponse);
+}
+
+message LoginRequest {
+  uint32 code = 1;
+  uint32 replicaIndex = 2;
+}
+
+message LoginResponse {
+  bool success = 1;
+}

api/grpc/source-telegram/service_logging.go

@@ -0,0 +1,30 @@
+package source_telegram
+
+import (
+	"context"
+	"fmt"
+	"log/slog"
+)
+
+type serviceLogging struct {
+	svc Service
+	log *slog.Logger
+}
+
+func NewServiceLogging(svc Service, log *slog.Logger) Service {
+	return serviceLogging{
+		svc: svc,
+		log: log,
+	}
+}
+
+func (sl serviceLogging) Login(ctx context.Context, code int64, replicaIdx uint) (err error) {
+	err = sl.svc.Login(ctx, code, replicaIdx)
+	switch err {
+	case nil:
+		sl.log.Debug(fmt.Sprintf("api.grpc.source-telegram.Login(%d. %d): ok", code, replicaIdx))
+	default:
+		sl.log.Error(fmt.Sprintf("api.grpc.source-telegram.Login(%d, %d): %s", code, replicaIdx, err))
+	}
+	return
+}

config/config.go

@@ -27,9 +27,10 @@ type Config struct {
 			Token                       string `envconfig:"API_TELEGRAM_TOKEN" required:"true"`
 			PublicInterestChannelPrefix string `envconfig:"API_TELEGRAM_PUBLIC_INTEREST_CHANNEL_PREFIX" default:"awk_" required:"true"`
 		}
-		Uri    string `envconfig:"API_URI" default:"api:50051" required:"true"`
-		Reader ReaderConfig
-		Queue  QueueConfig
+		Uri            string `envconfig:"API_URI" default:"api:50051" required:"true"`
+		Reader         ReaderConfig
+		Queue          QueueConfig
+		SourceTelegram SourceTelegramConfig
 	}
 	Payment PaymentConfig
 	Log     struct {
@@ -96,8 +97,12 @@ type MessagesConfig struct {
 	Uri  string `envconfig:"API_MESSAGES_URI" default:"messages:50051" required:"true"`
 }
 
+type SourceTelegramConfig struct {
+	Uri string `envconfig:"API_SOURCE_TELEGRAM_URI" default:"source-telegram:50051" required:"true"`
+}
+
 type LoginCodeConfig struct {
-	FromUserIds       map[int64]bool `envconfig:"LOGIN_CODE_FROM_USER_IDS" required:"true"`
+	FromUserIds       map[int64]uint `envconfig:"LOGIN_CODE_FROM_USER_IDS" required:"true"`
 	ForwardFromUserId int64          `envconfig:"LOGIN_CODE_FORWARD_FROM_USER_ID" required:"true" default:"777000"`
 }
 

config/config_test.go

@@ -15,15 +15,15 @@ func TestConfig(t *testing.T) {
 	os.Setenv("PAYMENT_PROVIDER_TOKEN", "yohoho")
 	os.Setenv("REPLICA_RANGE", "2")
 	os.Setenv("REPLICA_NAME", "replica-0")
-	os.Setenv("LOGIN_CODE_FROM_USER_IDS", "123:true,456:true")
+	os.Setenv("LOGIN_CODE_FROM_USER_IDS", "123:0,456:1")
 	cfg, err := NewConfigFromEnv()
 	assert.Nil(t, err)
 	assert.Equal(t, uint16(56789), cfg.Api.Telegram.Webhook.Port)
 	assert.Equal(t, uint16(45678), cfg.Api.Telegram.Bot.Port)
 	assert.Equal(t, 4, cfg.Log.Level)
 	assert.Equal(t, "yohoho", cfg.Payment.Provider.Token)
-	assert.Equal(t, map[int64]bool{
-		123: true,
-		456: true,
+	assert.Equal(t, map[int64]uint{
+		123: 0,
+		456: 1,
 	}, cfg.LoginCode.FromUserIds)
 }

helm/bot-telegram/templates/deployment.yaml

@@ -108,6 +108,8 @@ spec:
               value: "{{ .Values.api.queue.interestsCreated.name }}"
             - name: API_QUEUE_INTERESTS_CREATED_SUBJ
               value: "{{ .Values.api.queue.interestsCreated.subj }}"
+            - name: API_SOURCE_TELEGRAM_URI
+              value: "{{ .Values.api.sourceTelegram.uri }}"
             - name: LOGIN_CODE_FROM_USER_IDS
               valueFrom:
                 secretKeyRef:

helm/bot-telegram/values.yaml

@@ -120,6 +120,8 @@ api:
       batchSize: 10
       name: "bot-telegram"
       subj: "interests-created"
+  sourceTelegram:
+    uri: "source-telegram:50051"
 cert:
   acme:
     email: "awakari@awakari.com"

main.go

@@ -8,6 +8,7 @@ import (
 	grpcApiAdmin "github.com/awakari/bot-telegram/api/grpc/admin"
 	grpcApiMsgs "github.com/awakari/bot-telegram/api/grpc/messages"
 	"github.com/awakari/bot-telegram/api/grpc/queue"
+	apiGrpcSrcTg "github.com/awakari/bot-telegram/api/grpc/source-telegram"
 	grpcApiTgBot "github.com/awakari/bot-telegram/api/grpc/tgbot"
 	"github.com/awakari/bot-telegram/api/http/reader"
 	"github.com/awakari/bot-telegram/config"
@@ -123,6 +124,9 @@ func main() {
 		}
 	}()
 
+	svcSrcTg := apiGrpcSrcTg.NewService(cfg.Api.SourceTelegram.Uri)
+	svcSrcTg = apiGrpcSrcTg.NewServiceLogging(svcSrcTg, log)
+
 	// init events format, see https://core.telegram.org/bots/api#html-style for details
 	htmlPolicy := bluemonday.NewPolicy()
 	htmlPolicy.AllowStandardURLs()
@@ -200,6 +204,7 @@ func main() {
 	fwdHandler := service.LoginCodeHandler{
 		FromUserIds:  cfg.LoginCode.FromUserIds,
 		SourceUserId: cfg.LoginCode.ForwardFromUserId,
+		SvcSrcTg:     svcSrcTg,
 	}
 	txtHandlers := map[string]telebot.HandlerFunc{}
 	hRoot := service.RootHandler{

service/login.go

@@ -1,16 +1,19 @@
 package service
 
 import (
+	"context"
 	"errors"
 	"fmt"
+	apiGrpcSrcTg "github.com/awakari/bot-telegram/api/grpc/source-telegram"
 	"gopkg.in/telebot.v3"
 	"regexp"
 	"strconv"
 )
 
 type LoginCodeHandler struct {
-	FromUserIds  map[int64]bool
+	FromUserIds  map[int64]uint
 	SourceUserId int64
+	SvcSrcTg     apiGrpcSrcTg.Service
 }
 
 var rLoginCode = regexp.MustCompile(`Login code: (\d+).*`)
@@ -24,7 +27,8 @@ func (h LoginCodeHandler) Handle(tgCtx telebot.Context) (err error) {
 		return
 	}
 	userId := msg.Sender.ID
-	if !h.FromUserIds[userId] {
+	replicaIdx, replicaIdxPresent := h.FromUserIds[userId]
+	if !replicaIdxPresent {
 		err = fmt.Errorf("message is forwarded by user %d", userId)
 		return
 	}
@@ -33,7 +37,7 @@ func (h LoginCodeHandler) Handle(tgCtx telebot.Context) (err error) {
 	if err != nil {
 		return
 	}
-	fmt.Printf("LOGIN CODE: %d\n", code)
+	err = h.SvcSrcTg.Login(context.TODO(), int64(code), replicaIdx)
 	return
 }