Adam Van Ymeren
7a04f298d2
- update to latest telegram layer - remove some references to fields in tg.Entities that don't exist in the schema - originally added here: https://github.com/beeper/td/commit/820929062a2ba0104397bc01235ab58a9cff780e - referenced here - https://github.com/mautrix/telegramgo/commit/124f0967ed195b5a380c9bd02e170ada9710dde3 - https://github.com/mautrix/telegramgo/commit/4205047aab2e0639217148b5d125bfaab668bd8e
105 lines
3.0 KiB
Go
105 lines
3.0 KiB
Go
package telegram
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"strings"
|
|
|
|
"go.opentelemetry.io/otel/attribute"
|
|
"go.opentelemetry.io/otel/trace"
|
|
"go.uber.org/zap"
|
|
|
|
"go.mau.fi/mautrix-telegram/pkg/gotd/bin"
|
|
"go.mau.fi/mautrix-telegram/pkg/gotd/tg"
|
|
"go.mau.fi/mautrix-telegram/pkg/gotd/tgerr"
|
|
)
|
|
|
|
// API returns *tg.Client for calling raw MTProto methods.
|
|
func (c *Client) API() *tg.Client {
|
|
return c.tg
|
|
}
|
|
|
|
// Invoke invokes raw MTProto RPC method. It sends input and decodes result
|
|
// into output.
|
|
func (c *Client) Invoke(ctx context.Context, input bin.Encoder, output bin.Decoder) error {
|
|
if c.tracer != nil {
|
|
spanName := "Invoke"
|
|
var attrs []attribute.KeyValue
|
|
if t, ok := input.(interface{ TypeID() uint32 }); ok {
|
|
id := t.TypeID()
|
|
attrs = append(attrs,
|
|
attribute.Int64("tg.method.id_int", int64(id)),
|
|
attribute.String("tg.method.id", fmt.Sprintf("%x", id)),
|
|
)
|
|
name := c.opts.Types.Get(id)
|
|
if name == "" {
|
|
name = fmt.Sprintf("0x%x", id)
|
|
} else {
|
|
attrs = append(attrs, attribute.String("tg.method.name", name))
|
|
}
|
|
spanName = fmt.Sprintf("Invoke: %s", name)
|
|
}
|
|
spanCtx, span := c.tracer.Start(ctx, spanName,
|
|
trace.WithAttributes(attrs...),
|
|
trace.WithSpanKind(trace.SpanKindClient),
|
|
)
|
|
ctx = spanCtx
|
|
defer span.End()
|
|
}
|
|
|
|
return c.invoker.Invoke(ctx, input, output)
|
|
}
|
|
|
|
// invokeDirect directly invokes RPC method, automatically handling datacenter redirects.
|
|
func (c *Client) invokeDirect(ctx context.Context, input bin.Encoder, output bin.Decoder) error {
|
|
contextDC, _ := ctx.Value("tg_dc").(*int)
|
|
|
|
if _, ok := ctx.Value("tg_dc_inner").(bool); !ok {
|
|
if contextDC != nil && *contextDC > 0 {
|
|
c.log.With(zap.Int("context_dc", *contextDC)).Debug("Invoking on context DC")
|
|
return c.invokeSub(ctx, *contextDC, input, output)
|
|
}
|
|
}
|
|
|
|
if err := c.invokeConn(ctx, input, output); err != nil {
|
|
// Handling datacenter migration request.
|
|
if rpcErr, ok := tgerr.As(err); ok && strings.HasSuffix(rpcErr.Type, "_MIGRATE") {
|
|
targetDC := rpcErr.Argument
|
|
log := c.log.With(
|
|
zap.String("error_type", rpcErr.Type),
|
|
zap.Int("target_dc", targetDC),
|
|
)
|
|
// If migration error is FILE_MIGRATE or STATS_MIGRATE, then the method
|
|
// called by authorized client, so we should try to transfer auth to new DC
|
|
// and create new connection.
|
|
if rpcErr.IsOneOf("FILE_MIGRATE", "STATS_MIGRATE") {
|
|
log.Debug("Invoking on target DC")
|
|
if contextDC != nil {
|
|
log.Debug("Setting context DC")
|
|
*contextDC = targetDC
|
|
ctx = context.WithValue(ctx, "tg_dc_inner", true)
|
|
}
|
|
return c.invokeSub(ctx, targetDC, input, output)
|
|
}
|
|
|
|
// Otherwise we should change primary DC.
|
|
log.Info("Migrating to target DC")
|
|
return c.invokeMigrate(ctx, targetDC, input, output)
|
|
}
|
|
|
|
return err
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
// invokeConn directly invokes RPC call on primary connection without any
|
|
// additional handling.
|
|
func (c *Client) invokeConn(ctx context.Context, input bin.Encoder, output bin.Decoder) error {
|
|
c.connMux.Lock()
|
|
conn := c.conn
|
|
c.connMux.Unlock()
|
|
|
|
return conn.Invoke(ctx, input, output)
|
|
}
|