diff --git a/pkg/connector/block_user.go b/pkg/connector/block_user.go index f783e426..8cae845c 100644 --- a/pkg/connector/block_user.go +++ b/pkg/connector/block_user.go @@ -26,7 +26,7 @@ func (lc *LineClient) HandleMatrixBlockUser(ctx context.Context, msg *bridgev2.M _, err := lc.callLine(ctx, func(client *line.Client) error { return client.SetContactBlocked(ctx, int64(reqSeq), mid, msg.Content.Block) }) - return err + return wrapSendError(err) } func (lc *LineClient) queueUserBlocked(ctx context.Context, mid string, blocked bool) bool { diff --git a/pkg/connector/client.go b/pkg/connector/client.go index 97113f3d..4369df38 100644 --- a/pkg/connector/client.go +++ b/pkg/connector/client.go @@ -26,6 +26,14 @@ var ( errLineQRLoginRequired = errors.New("LINE requires a new QR scan. Reconnect in Beeper to continue") ) +func init() { + status.BridgeStateHumanErrors.Update(status.BridgeStateErrorMap{ + "line-login-failed": "LINE could not sign you in. Sign in again.", + "line-token-expired": "Your LINE session expired and could not be restored. Sign in again.", + "line-sse-disconnected": "Disconnected from LINE, reconnecting", + }) +} + const lineMissingE2EEKeyMessage = "LINE encryption keys are unavailable. Reconnect LINE in Beeper to restore message decryption." var newLineAPIClient = line.NewClient @@ -409,6 +417,10 @@ func (lc *LineClient) sendConnectedStateIfCurrent(ctx context.Context) bool { // Whichever transition gets this lock last determines the visible state. lc.recoverMu.Lock() defer lc.recoverMu.Unlock() + return lc.sendConnectedStateLocked(ctx) +} + +func (lc *LineClient) sendConnectedStateLocked(ctx context.Context) bool { if ctx.Err() != nil || lc.superseded.Load() || lc.isSessionInvalidated() || !lc.hasAccessToken() { return false } @@ -419,6 +431,34 @@ func (lc *LineClient) sendConnectedStateIfCurrent(ctx context.Context) bool { return true } +// SSE reconnects must not replace the relogin prompt sent by markMissingE2EEKey. +func (lc *LineClient) sendSSEConnectedIfCurrent(ctx context.Context) bool { + lc.recoverMu.Lock() + defer lc.recoverMu.Unlock() + lc.missingE2EEKeyMu.Lock() + defer lc.missingE2EEKeyMu.Unlock() + if lc.missingE2EEKeyMarked { + return false + } + return lc.sendConnectedStateLocked(ctx) +} + +func (lc *LineClient) sendSSEDisconnectedIfCurrent(ctx context.Context, err error) bool { + lc.recoverMu.Lock() + defer lc.recoverMu.Unlock() + lc.missingE2EEKeyMu.Lock() + defer lc.missingE2EEKeyMu.Unlock() + if ctx.Err() != nil || lc.superseded.Load() || lc.isSessionInvalidated() || lc.missingE2EEKeyMarked || lc.UserLogin.BridgeState == nil { + return false + } + lc.UserLogin.BridgeState.Send(status.BridgeState{ + StateEvent: status.StateTransientDisconnect, + Error: "line-sse-disconnected", + Info: map[string]any{"go_error": err.Error()}, + }) + return true +} + func (lc *LineClient) saveSessionInvalidated(ctx context.Context) { if lc.UserLogin == nil || lc.UserLogin.UserLogin == nil { return @@ -603,7 +643,7 @@ func (lc *LineClient) Connect(ctx context.Context) { lc.UserLogin.BridgeState.Send(status.BridgeState{ StateEvent: status.StateBadCredentials, Error: "line-login-failed", - Message: err.Error(), + Info: map[string]any{"go_error": err.Error()}, UserAction: status.UserActionRelogin, }) return @@ -622,7 +662,7 @@ func (lc *LineClient) Connect(ctx context.Context) { lc.UserLogin.BridgeState.Send(status.BridgeState{ StateEvent: status.StateBadCredentials, Error: "line-token-expired", - Message: fmt.Sprintf("session expired and could not be restored: %v", err), + Info: map[string]any{"go_error": err.Error()}, UserAction: status.UserActionRelogin, }) return diff --git a/pkg/connector/connector.go b/pkg/connector/connector.go index 9cb784df..ccf3e06f 100644 --- a/pkg/connector/connector.go +++ b/pkg/connector/connector.go @@ -724,7 +724,7 @@ func (ll *LineEmailLogin) finishLogin(ctx context.Context, res *line.LoginResult client.LoginAttempt = res.Attempt profile, err := client.GetProfileContext(ctx) if err != nil { - return nil, fmt.Errorf("failed to verify token: %w", err) + return nil, wrapLineLoginError(fmt.Errorf("failed to verify token: %w", err)) } displayName := profile.DisplayName diff --git a/pkg/connector/creategroup.go b/pkg/connector/creategroup.go index dcbbd70c..d61dc643 100644 --- a/pkg/connector/creategroup.go +++ b/pkg/connector/creategroup.go @@ -39,7 +39,7 @@ func (lc *LineClient) CreateGroup(ctx context.Context, params *bridgev2.GroupCre return client.CreateChat(participantMids, lineName, chatType) }) if err != nil { - return nil, fmt.Errorf("failed to create LINE chat: %w", err) + return nil, wrapLineRespError(fmt.Errorf("failed to create LINE chat: %w", err)) } lc.UserLogin.Bridge.Log.Info(). diff --git a/pkg/connector/delete_chat.go b/pkg/connector/delete_chat.go index a7b35dcc..2cb12b0a 100644 --- a/pkg/connector/delete_chat.go +++ b/pkg/connector/delete_chat.go @@ -23,13 +23,13 @@ func (lc *LineClient) HandleMatrixDeleteChat(ctx context.Context, msg *bridgev2. _, err := lc.callLine(ctx, func(client *line.Client) error { return client.RejectChatInvitationContext(ctx, int64(reqSeq), chatMid) }) - return err + return wrapSendError(err) } _, boxes, err := callLineResult(lc, ctx, func(client *line.Client) (map[string]line.MessageBox, error) { return client.GetMessageBoxesByIDsContext(ctx, []string{chatMid}) }) if err != nil { - return err + return wrapSendError(err) } lastDelivered := boxes[chatMid].LastDeliveredMessageID if lastDelivered == nil || lastDelivered.MessageID == "" || lastDelivered.MessageID == "-1" { @@ -43,7 +43,7 @@ func (lc *LineClient) HandleMatrixDeleteChat(ctx context.Context, msg *bridgev2. return client.SendChatRemovedContext(ctx, int64(reqSeq), chatMid, lastDelivered.MessageID, 0) }) if err != nil { - return err + return wrapSendError(err) } return lc.saveChatDeletion(context.WithoutCancel(ctx), chatMid, lastDelivered.MessageID) } diff --git a/pkg/connector/login_qr.go b/pkg/connector/login_qr.go index ac998ed9..eb1b1471 100644 --- a/pkg/connector/login_qr.go +++ b/pkg/connector/login_qr.go @@ -49,6 +49,7 @@ func (lq *LineQRLogin) Start(ctx context.Context) (*bridgev2.LoginStep, error) { defer stop() client := newLineAPIClient("") session, qr, err := client.StartQRLogin(processCtx) + err = wrapLineLoginError(err) ll.mu.Lock() ll.attempt = client.LoginAttempt canceled := ll.canceled || ctx.Err() != nil || processCtx.Err() != nil @@ -123,7 +124,7 @@ func (lq *LineQRLogin) Wait(ctx context.Context) (*bridgev2.LoginStep, error) { verified, err := lq.client.VerifyQRCertificate(ctx, lq.session, lq.login.Certificate) if err != nil { lq.Cancel() - return nil, err + return nil, wrapLineLoginError(err) } if verified { return lq.finish(ctx) @@ -131,7 +132,7 @@ func (lq *LineQRLogin) Wait(ctx context.Context) (*bridgev2.LoginStep, error) { pin, err := lq.client.CreatePinCode(ctx, lq.session) if err != nil { lq.Cancel() - return nil, err + return nil, wrapLineLoginError(err) } lq.pin = true lq.startPoll(lq.client.CheckPinCodeVerifiedContext, false) @@ -148,7 +149,7 @@ func (lq *LineQRLogin) finish(ctx context.Context) (*bridgev2.LoginStep, error) defer lq.Cancel() res, err := lq.client.QRCodeLoginV2(ctx, lq.session) if err != nil { - return nil, err + return nil, wrapLineLoginError(err) } return lq.login.finishLogin(ctx, res) } diff --git a/pkg/connector/loginerrors.go b/pkg/connector/loginerrors.go index 18760c16..2f9bd1fa 100644 --- a/pkg/connector/loginerrors.go +++ b/pkg/connector/loginerrors.go @@ -1,10 +1,12 @@ package connector import ( + "context" "errors" "fmt" "net/http" + "maunium.net/go/mautrix" "maunium.net/go/mautrix/bridgev2" "github.com/beeper/line/pkg/line" @@ -29,18 +31,14 @@ var ( ErrLoginTooManyAttempts = bridgev2.RespError{ ErrCode: "DEV.HIGHEST.LINE.TOO_MANY_ATTEMPTS", Err: loginTooManyAttemptsReason, - StatusCode: http.StatusTooManyRequests, + StatusCode: http.StatusBadRequest, } ErrLoginRejected = bridgev2.RespError{ ErrCode: "DEV.HIGHEST.LINE.LOGIN_REJECTED", Err: genericLoginFailureReason, StatusCode: http.StatusUnauthorized, } - ErrLoginUnknown = bridgev2.RespError{ - ErrCode: "M_UNKNOWN", - Err: "Internal error logging in to LINE", - StatusCode: http.StatusInternalServerError, - } + ErrLoginUnknown = bridgev2.RespError(mautrix.MUnknown.WithMessage("Internal error logging in to LINE")) ) // wrapLineLoginError translates a LINE error into one the client can act on, keeping the @@ -70,3 +68,33 @@ func wrapLineLoginError(err error) error { } return fmt.Errorf("%w: %w", mapped, err) } + +// wrapLineRespError translates an error from a LINE call made for a provisioning +// request into an HTTP error, keeping the original error in the chain. +func wrapLineRespError(err error) error { + if err == nil { + return nil + } + var respErr bridgev2.RespError + if errors.As(err, &respErr) { + return err + } + var mapped bridgev2.RespError + switch statusCode := line.HTTPStatusCode(err); { + case errors.Is(err, errLineSessionInvalidated), errors.Is(err, errLineClientSuperseded), line.IsAuthError(err): + mapped = bridgev2.RespError(mautrix.MForbidden.WithMessage("Not logged in to LINE. Reconnect and try again.")) + case line.IsNotAMemberError(err): + mapped = bridgev2.RespError(mautrix.MForbidden.WithMessage("You are not a member of this LINE chat")) + case line.IsTalkExceptionNotFound(err): + mapped = bridgev2.RespError(mautrix.MNotFound.WithMessage("Not found on LINE")) + case statusCode == http.StatusTooManyRequests: + mapped = bridgev2.RespError(mautrix.MLimitExceeded.WithMessage("LINE is rate limiting requests. Please wait a moment and try again.").WithStatus(http.StatusBadRequest)) + case errors.Is(err, context.DeadlineExceeded): + mapped = bridgev2.RespError(mautrix.MUnknown.WithMessage("Timed out waiting for LINE")) + case line.IsTalkException(err), statusCode >= 400 && statusCode < 500: + mapped = bridgev2.RespError(mautrix.MForbidden.WithMessage("LINE rejected the request")) + default: + mapped = bridgev2.RespError(mautrix.MUnknown.WithMessage("Request to LINE failed")) + } + return fmt.Errorf("%w: %w", mapped, err) +} diff --git a/pkg/connector/reaction.go b/pkg/connector/reaction.go index a701e7cf..f179716e 100644 --- a/pkg/connector/reaction.go +++ b/pkg/connector/reaction.go @@ -607,14 +607,6 @@ func invalidReactionTargetError(messageID string) error { WithErrorReason(event.MessageStatusUnsupported) } -func reactionNotAMemberError() error { - return bridgev2.WrapErrorInStatus(fmt.Errorf("LINE says this account is not a member of the chat")). - WithStatus(event.MessageStatusFail). - WithIsCertain(true). - WithErrorAsMessage(). - WithErrorReason(event.MessageStatusNoPermission) -} - func parseReactionTargetMessageID(messageID networkid.MessageID) (string, error) { raw := string(messageID) if raw == "" || strings.HasPrefix(raw, "local-") || strings.HasPrefix(raw, "$") { @@ -737,10 +729,7 @@ func (lc *LineClient) HandleMatrixReaction(ctx context.Context, msg *bridgev2.Ma if line.IsInvalidPaidReactionType(err) { return nil, unsupportedMatrixReactionError(key) } - if line.IsNotAMemberError(err) { - return nil, reactionNotAMemberError() - } - return nil, err + return nil, wrapSendError(err) } return &database.Reaction{ @@ -762,8 +751,5 @@ func (lc *LineClient) HandleMatrixReactionRemove(ctx context.Context, msg *bridg _, err = lc.callLine(ctx, func(client *line.Client) error { return client.CancelReaction(int64(reqSeq), targetID) }) - if line.IsNotAMemberError(err) { - return reactionNotAMemberError() - } - return err + return wrapSendError(err) } diff --git a/pkg/connector/reaction_test.go b/pkg/connector/reaction_test.go index a34ff15a..6d5efe58 100644 --- a/pkg/connector/reaction_test.go +++ b/pkg/connector/reaction_test.go @@ -632,7 +632,7 @@ func TestUnsupportedMatrixReactionError(t *testing.T) { } func TestReactionNotAMemberError(t *testing.T) { - err := reactionNotAMemberError() + err := wrapSendError(errors.New(`API error 400: {"code":10051,"message":"RESPONSE_ERROR","data":{"name":"TalkException","code":10,"reason":"not a member"}}`)) var status bridgev2.MessageStatus if !errors.As(err, &status) { t.Fatalf("error %T does not wrap bridgev2.MessageStatus", err) @@ -643,8 +643,8 @@ func TestReactionNotAMemberError(t *testing.T) { if status.ErrorReason != event.MessageStatusNoPermission { t.Fatalf("ErrorReason = %q, want %q", status.ErrorReason, event.MessageStatusNoPermission) } - if !status.IsCertain || !status.ErrorAsMessage { - t.Fatalf("status certainty/message flags = %v/%v, want true/true", status.IsCertain, status.ErrorAsMessage) + if !status.IsCertain || status.Message == "" { + t.Fatalf("status certainty/message = %v/%q, want true and a message", status.IsCertain, status.Message) } } diff --git a/pkg/connector/send_message.go b/pkg/connector/send_message.go index 9632ad22..d5031282 100644 --- a/pkg/connector/send_message.go +++ b/pkg/connector/send_message.go @@ -59,7 +59,7 @@ func lineGroupE2EEFetchFailureError(err error) error { if errStatus := lineGroupE2EEReconnectRequiredError(err); errStatus != nil { return errStatus } - return err + return wrapSendError(err) } func (lc *LineClient) HandleMatrixMessage(ctx context.Context, msg *bridgev2.MatrixMessage) (*bridgev2.MatrixMessageResponse, error) { @@ -94,10 +94,7 @@ func (lc *LineClient) HandleMatrixMessage(ctx context.Context, msg *bridgev2.Mat // Block sends to blocked contacts in DMs. if !isGroup && lc.isUserBlocked(portalMid) { - return nil, bridgev2.WrapErrorInStatus(fmt.Errorf("user is blocked on LINE: unblock them to send messages")). - WithIsCertain(true). - WithSendNotice(true). - WithErrorReason(event.MessageStatusGenericError) + return nil, errBlockedContact } // Determine whether we need to send as plain text (peer/group has Letter Sealing off). @@ -115,7 +112,7 @@ func (lc *LineClient) HandleMatrixMessage(ctx context.Context, msg *bridgev2.Mat if errPeer != nil && line.IsNoUsableE2EEPublicKey(errPeer) { plainText = true } else if errPeer != nil { - return nil, fmt.Errorf("failed to get peer key: %w", errPeer) + return nil, wrapSendError(fmt.Errorf("failed to get peer key: %w", errPeer)) } } @@ -183,14 +180,14 @@ func (lc *LineClient) HandleMatrixMessage(ctx context.Context, msg *bridgev2.Mat } else { payload, err = json.Marshal(map[string]string{"text": msg.Content.Body}) if err != nil { - return nil, fmt.Errorf("failed to marshal text payload: %w", err) + return nil, wrapConvertError(fmt.Errorf("failed to marshal text payload: %w", err)) } } case event.MsgImage: data, err := lc.UserLogin.Bridge.Bot.DownloadMedia(ctx, msg.Content.URL, msg.Content.File) if err != nil { - return nil, fmt.Errorf("failed to download media from matrix: %w", err) + return nil, fmt.Errorf("%w: %w", bridgev2.ErrMediaDownloadFailed, err) } mimeType := msg.Content.Info.MimeType @@ -198,7 +195,7 @@ func (lc *LineClient) HandleMatrixMessage(ctx context.Context, msg *bridgev2.Mat if isHEICImage(mimeType, fileName) { data, err = convertHEICToJPEG(ctx, data) if err != nil { - return nil, fmt.Errorf("failed to convert HEIC image to JPEG: %w", err) + return nil, fmt.Errorf("%w: %w", bridgev2.ErrMediaConvertFailed, err) } mimeType = "image/jpeg" fileName = jpegFileName(fileName) @@ -264,14 +261,14 @@ func (lc *LineClient) HandleMatrixMessage(ctx context.Context, msg *bridgev2.Mat uploadData, keyMaterialB64, err := lc.encryptFileData(data) if err != nil { - return nil, fmt.Errorf("failed to encrypt image data: %w", err) + return nil, wrapConvertError(fmt.Errorf("failed to encrypt image data: %w", err)) } oid, err := callLineString(func(client *line.Client) (string, error) { return client.UploadOBS(uploadData) }) if err != nil { - return nil, fmt.Errorf("failed to upload image to OBS: %w", err) + return nil, wrapMediaUploadError(fmt.Errorf("failed to upload image to OBS: %w", err)) } thumbnailData, thumbWidth, thumbHeight, err := generateThumbnail(data) @@ -333,7 +330,7 @@ func (lc *LineClient) HandleMatrixMessage(ctx context.Context, msg *bridgev2.Mat case event.MsgFile: data, err := lc.UserLogin.Bridge.Bot.DownloadMedia(ctx, msg.Content.URL, msg.Content.File) if err != nil { - return nil, fmt.Errorf("failed to download file from matrix: %w", err) + return nil, fmt.Errorf("%w: %w", bridgev2.ErrMediaDownloadFailed, err) } contentType = int(ContentFile) @@ -365,14 +362,14 @@ func (lc *LineClient) HandleMatrixMessage(ctx context.Context, msg *bridgev2.Mat uploadData, keyMaterialB64, err := lc.encryptFileData(data) if err != nil { - return nil, fmt.Errorf("failed to encrypt file data: %w", err) + return nil, wrapConvertError(fmt.Errorf("failed to encrypt file data: %w", err)) } oid, err := callLineString(func(client *line.Client) (string, error) { return client.UploadOBSWithSID(uploadData, "emf") }) if err != nil { - return nil, fmt.Errorf("failed to upload file to OBS: %w", err) + return nil, wrapMediaUploadError(fmt.Errorf("failed to upload file to OBS: %w", err)) } contentMetadata["OID"] = oid @@ -398,7 +395,7 @@ func (lc *LineClient) HandleMatrixMessage(ctx context.Context, msg *bridgev2.Mat case event.MsgVideo: data, err := lc.UserLogin.Bridge.Bot.DownloadMedia(ctx, msg.Content.URL, msg.Content.File) if err != nil { - return nil, fmt.Errorf("failed to download video from matrix: %w", err) + return nil, fmt.Errorf("%w: %w", bridgev2.ErrMediaDownloadFailed, err) } contentType = int(ContentVideo) @@ -448,14 +445,14 @@ func (lc *LineClient) HandleMatrixMessage(ctx context.Context, msg *bridgev2.Mat uploadData, keyMaterialB64, err := lc.encryptVideoData(data) if err != nil { - return nil, fmt.Errorf("failed to encrypt video data: %w", err) + return nil, wrapConvertError(fmt.Errorf("failed to encrypt video data: %w", err)) } oid, err := callLineString(func(client *line.Client) (string, error) { return client.UploadOBSWithSID(uploadData, "emv") }) if err != nil { - return nil, fmt.Errorf("failed to upload video to OBS: %w", err) + return nil, wrapMediaUploadError(fmt.Errorf("failed to upload video to OBS: %w", err)) } chunkHashes := generateChunkHashes(uploadData[:len(uploadData)-32]) @@ -529,7 +526,7 @@ func (lc *LineClient) HandleMatrixMessage(ctx context.Context, msg *bridgev2.Mat case event.MsgAudio: data, err := lc.UserLogin.Bridge.Bot.DownloadMedia(ctx, msg.Content.URL, msg.Content.File) if err != nil { - return nil, fmt.Errorf("failed to download audio from matrix: %w", err) + return nil, fmt.Errorf("%w: %w", bridgev2.ErrMediaDownloadFailed, err) } contentType = int(ContentAudio) @@ -550,14 +547,14 @@ func (lc *LineClient) HandleMatrixMessage(ctx context.Context, msg *bridgev2.Mat uploadData, keyMaterialB64, err := lc.encryptFileData(data) if err != nil { - return nil, fmt.Errorf("failed to encrypt audio data: %w", err) + return nil, wrapConvertError(fmt.Errorf("failed to encrypt audio data: %w", err)) } oid, err := callLineString(func(client *line.Client) (string, error) { return client.UploadOBSWithSID(uploadData, "ema") }) if err != nil { - return nil, fmt.Errorf("failed to upload audio to OBS: %w", err) + return nil, wrapMediaUploadError(fmt.Errorf("failed to upload audio to OBS: %w", err)) } contentMetadata["OID"] = oid @@ -574,7 +571,7 @@ func (lc *LineClient) HandleMatrixMessage(ctx context.Context, msg *bridgev2.Mat } default: - return nil, fmt.Errorf("message type %s not implemented", effectiveMsgType) + return nil, fmt.Errorf("%w %s", bridgev2.ErrUnsupportedMessageType, effectiveMsgType) } // Process mentions — set MENTION content metadata for LINE notification @@ -652,18 +649,18 @@ func (lc *LineClient) HandleMatrixMessage(ctx context.Context, msg *bridgev2.Mat // 1-1 Encryption (peer key already fetched above) myRaw, myKeyID, errKey := lc.E2EE.MyKeyIDs() if errKey != nil { - return nil, fmt.Errorf("missing own E2EE key: %w", errKey) + return nil, wrapConvertError(fmt.Errorf("missing own E2EE key: %w", errKey)) } peerRaw, peerPub, errPeer := lc.ensurePeerKey(ctx, portalMid) if errPeer != nil { - return nil, fmt.Errorf("failed to get peer key: %w", errPeer) + return nil, wrapSendError(fmt.Errorf("failed to get peer key: %w", errPeer)) } chunks, err = lc.E2EE.EncryptMessageV2Raw(portalMid, fromMid, myKeyID, peerPub, myRaw, peerRaw, contentType, payload) } if err != nil { - return nil, fmt.Errorf("encrypt failed: %w", err) + return nil, wrapConvertError(fmt.Errorf("encrypt failed: %w", err)) } } @@ -722,7 +719,7 @@ func (lc *LineClient) HandleMatrixMessage(ctx context.Context, msg *bridgev2.Mat zipData, zipErr := wrapInZip(rawFileName, rawFileData) if zipErr != nil { - return nil, fmt.Errorf("failed to wrap file in zip: %w", zipErr) + return nil, wrapConvertError(fmt.Errorf("failed to wrap file in zip: %w", zipErr)) } zipFileName := rawFileName + ".zip" @@ -736,13 +733,13 @@ func (lc *LineClient) HandleMatrixMessage(ctx context.Context, msg *bridgev2.Mat // Re-encrypt and re-upload the zipped data uploadData, keyMaterialB64, encErr := lc.encryptFileData(zipData) if encErr != nil { - return nil, fmt.Errorf("failed to encrypt zipped file: %w", encErr) + return nil, wrapConvertError(fmt.Errorf("failed to encrypt zipped file: %w", encErr)) } oid, uploadErr := callLineString(func(client *line.Client) (string, error) { return client.UploadOBSWithSID(uploadData, "emf") }) if uploadErr != nil { - return nil, fmt.Errorf("failed to upload zipped file to OBS: %w", uploadErr) + return nil, wrapMediaUploadError(fmt.Errorf("failed to upload zipped file to OBS: %w", uploadErr)) } contentMetadata["OID"] = oid contentMetadata["ENC_KM"] = keyMaterialB64 @@ -767,16 +764,16 @@ func (lc *LineClient) HandleMatrixMessage(ctx context.Context, msg *bridgev2.Mat } else { myRaw, myKeyID, errKey := lc.E2EE.MyKeyIDs() if errKey != nil { - return nil, fmt.Errorf("missing own E2EE key for zip retry: %w", errKey) + return nil, wrapConvertError(fmt.Errorf("missing own E2EE key for zip retry: %w", errKey)) } peerRaw, peerPub, errPeer := lc.ensurePeerKey(ctx, portalMid) if errPeer != nil { - return nil, fmt.Errorf("failed to get peer key for zip retry: %w", errPeer) + return nil, wrapSendError(fmt.Errorf("failed to get peer key for zip retry: %w", errPeer)) } chunks, err = lc.E2EE.EncryptMessageV2Raw(portalMid, fromMid, myKeyID, peerPub, myRaw, peerRaw, contentType, zipPayload) } if err != nil { - return nil, fmt.Errorf("failed to encrypt zipped file message: %w", err) + return nil, wrapConvertError(fmt.Errorf("failed to encrypt zipped file message: %w", err)) } lineMsg.Chunks = chunks } @@ -834,7 +831,7 @@ func (lc *LineClient) HandleMatrixMessage(ctx context.Context, msg *bridgev2.Mat } if err != nil { - return nil, err + return nil, wrapSendError(err) } // For plain media: upload media to r/talk/m/{serverMessageId} after sending @@ -852,7 +849,11 @@ func (lc *LineClient) HandleMatrixMessage(ctx context.Context, msg *bridgev2.Mat if err := callLineErr(func(client *line.Client) error { return client.UploadOBSPlain(plainMediaData, sentMsg.ID, obsType) }); err != nil { - return nil, fmt.Errorf("failed to upload plain media to OBS: %w", err) + return nil, bridgev2.WrapErrorInStatus(fmt.Errorf("failed to upload plain media to OBS: %w", err)). + WithIsCertain(false). + WithSendNotice(true). + WithStatus(event.MessageStatusFail). + WithMessage("the message was sent to LINE, but the attachment failed to upload") } lc.UserLogin.Bridge.Log.Info(). Str("message_id", sentMsg.ID). @@ -935,13 +936,22 @@ func (lc *LineClient) HandleMatrixMessageRemove(ctx context.Context, msg *bridge return client.UnsendMessage(int64(reqSeq), string(msg.TargetMessage.ID)) }) if err != nil && strings.Contains(err.Error(), "message too old") { - return bridgev2.WrapErrorInStatus(fmt.Errorf("message too old to unsend on LINE (24h limit)")). + return bridgev2.WrapErrorInStatus(fmt.Errorf("message too old to unsend on LINE (24h limit): %w", err)). WithStatus(event.MessageStatusFail). WithErrorReason(event.MessageStatusTooOld). WithIsCertain(true). - WithSendNotice(true) + WithSendNotice(true). + WithMessage("messages can only be unsent on LINE within 24 hours") } - return err + if line.IsTalkExceptionNotFound(err) { + return bridgev2.WrapErrorInStatus(err). + WithStatus(event.MessageStatusFail). + WithErrorReason(event.MessageStatusUnsupported). + WithIsCertain(true). + WithSendNotice(true). + WithMessage("the message was not found on LINE") + } + return wrapSendError(err) } func (lc *LineClient) HandleMatrixLeaveRoom(ctx context.Context, portal *bridgev2.Portal) error { @@ -955,7 +965,7 @@ func (lc *LineClient) HandleMatrixLeaveRoom(ctx context.Context, portal *bridgev _, err := lc.callLine(ctx, func(client *line.Client) error { return client.RejectChatInvitation(int64(reqSeq), string(portal.ID)) }) - return err + return wrapSendError(err) } lc.trackReqSeq(reqSeq) @@ -963,7 +973,7 @@ func (lc *LineClient) HandleMatrixLeaveRoom(ctx context.Context, portal *bridgev _, err := lc.callLine(ctx, func(client *line.Client) error { return client.SendChatRemoved(int64(reqSeq), string(portal.ID), "0", 0) }) - return err + return wrapSendError(err) } // Compile-time assertion that LineClient handles Beeper message-request acceptance. @@ -976,7 +986,7 @@ func (lc *LineClient) HandleMatrixAcceptMessageRequest(ctx context.Context, msg _, err := lc.callLine(ctx, func(client *line.Client) error { return client.AcceptChatInvitation(reqSeq, string(msg.Portal.ID)) }) - return err + return wrapSendError(err) } func (lc *LineClient) buildMentionMetadata(ctx context.Context, body, formattedBody string, mentions *event.Mentions) map[string]string { diff --git a/pkg/connector/senderrors.go b/pkg/connector/senderrors.go new file mode 100644 index 00000000..11b099e5 --- /dev/null +++ b/pkg/connector/senderrors.go @@ -0,0 +1,113 @@ +package connector + +import ( + "context" + "errors" + "fmt" + "net/http" + + "maunium.net/go/mautrix/bridgev2" + "maunium.net/go/mautrix/event" + + "github.com/beeper/line/pkg/line" +) + +var errBlockedContact = bridgev2.WrapErrorInStatus(errors.New("contact is blocked on LINE")). + WithIsCertain(true). + WithSendNotice(true). + WithErrorReason(event.MessageStatusNoPermission). + WithStatus(event.MessageStatusFail). + WithMessage("you blocked this contact on LINE, unblock them to send messages") + +func notAMemberStatus(err error) bridgev2.MessageStatus { + return bridgev2.WrapErrorInStatus(err). + WithIsCertain(true). + WithSendNotice(true). + WithErrorReason(event.MessageStatusNoPermission). + WithStatus(event.MessageStatusFail). + WithMessage("you are no longer a member of this LINE chat") +} + +// wrapSendError turns an error from a LINE call made for a Matrix event into a +// message status. Errors that already carry a status are returned unchanged. +func wrapSendError(err error) error { + if err == nil { + return nil + } + var ms bridgev2.MessageStatus + if errors.As(err, &ms) { + return err + } + switch statusCode := line.HTTPStatusCode(err); { + case errors.Is(err, errLineSessionInvalidated), errors.Is(err, errLineClientSuperseded), line.IsAuthError(err): + return bridgev2.WrapErrorInStatus(err). + WithIsCertain(true). + WithSendNotice(true). + WithMessage("not logged in to LINE, reconnect and try again") + case line.IsNotAMemberError(err): + return notAMemberStatus(err) + case errors.Is(err, context.DeadlineExceeded), errors.Is(err, context.Canceled): + return bridgev2.WrapErrorInStatus(err). + WithIsCertain(false). + WithSendNotice(true). + WithMessage("timed out waiting for LINE") + case statusCode == http.StatusTooManyRequests: + return bridgev2.WrapErrorInStatus(err). + WithIsCertain(true). + WithSendNotice(true). + WithErrorReason(event.MessageStatusNetworkError). + WithMessage("LINE is rate limiting requests, try again later") + case statusCode >= 500: + return bridgev2.WrapErrorInStatus(err). + WithIsCertain(false). + WithSendNotice(true). + WithErrorReason(event.MessageStatusNetworkError). + WithMessage("LINE server error") + case line.IsTalkException(err), statusCode >= 400: + return bridgev2.WrapErrorInStatus(err). + WithIsCertain(true). + WithSendNotice(true). + WithErrorReason(event.MessageStatusNetworkError). + WithMessage("LINE rejected the request") + default: + return bridgev2.WrapErrorInStatus(err). + WithIsCertain(false). + WithSendNotice(true). + WithMessage("failed to reach LINE") + } +} + +// wrapConvertError marks a failure to build the LINE message as certain, since +// nothing has been sent yet. +func wrapConvertError(err error) error { + if err == nil { + return nil + } + var ms bridgev2.MessageStatus + if errors.As(err, &ms) { + return err + } + return bridgev2.WrapErrorInStatus(err). + WithIsCertain(true). + WithSendNotice(true). + WithMessage("failed to prepare the message for LINE") +} + +// wrapMediaUploadError maps a failed OBS upload done before the message is +// sent, so the message is known not to have been delivered. +func wrapMediaUploadError(err error) error { + if err == nil { + return nil + } + var ms bridgev2.MessageStatus + if errors.As(err, &ms) { + return err + } + if line.HTTPStatusCode(err) == http.StatusRequestEntityTooLarge { + return fmt.Errorf("%w: %w", bridgev2.ErrMediaTooLarge, err) + } + if line.IsAuthError(err) || errors.Is(err, errLineSessionInvalidated) || errors.Is(err, errLineClientSuperseded) { + return wrapSendError(err) + } + return fmt.Errorf("%w: %w", bridgev2.ErrMediaReuploadFailed, err) +} diff --git a/pkg/connector/sync.go b/pkg/connector/sync.go index ad50eca7..c74ad6c8 100644 --- a/pkg/connector/sync.go +++ b/pkg/connector/sync.go @@ -1638,9 +1638,14 @@ func (lc *LineClient) pollLoop(ctx context.Context) { return false } + sseDisconnected := false handler := func(eventType, data string) { + if sseDisconnected && lc.sendSSEConnectedIfCurrent(ctx) { + sseDisconnected = false + } + // handle keep alives - if eventType == "ping" || eventType == "connInfoRevision" { + if eventType == line.SSEEventOpen || eventType == "ping" || eventType == "connInfoRevision" { return } @@ -1733,6 +1738,8 @@ func (lc *LineClient) pollLoop(ctx context.Context) { if lc.handleReceiveAuthError(ctx, client, err) { return } + } else if !sseDisconnected && lc.sendSSEDisconnectedIfCurrent(ctx, err) { + sseDisconnected = true } } if !waitForSSEReconnect(receiveCtx) { diff --git a/pkg/line/client.go b/pkg/line/client.go index 8fb531d9..6740f1e3 100644 --- a/pkg/line/client.go +++ b/pkg/line/client.go @@ -477,7 +477,7 @@ func (c *Client) callRPCWithBaseURLContext(ctx context.Context, baseURL, service respBody, _ := io.ReadAll(resp.Body) if resp.StatusCode != 200 { - return nil, fmt.Errorf("API error %d: %s", resp.StatusCode, string(respBody)) + return nil, newHTTPError(resp.StatusCode, "API error %d: %s", resp.StatusCode, string(respBody)) } return respBody, nil @@ -562,7 +562,7 @@ func (c *Client) postWithHMAC(fullURL string, body []byte) ([]byte, error) { if resp.StatusCode != 200 { respBody, _ := io.ReadAll(resp.Body) - return nil, fmt.Errorf("HTTP %d: %s", resp.StatusCode, string(respBody)) + return nil, newHTTPError(resp.StatusCode, "HTTP %d: %s", resp.StatusCode, string(respBody)) } return io.ReadAll(resp.Body) @@ -635,7 +635,7 @@ func (c *Client) UploadOBSPlain(data []byte, oid string, obsType string) error { url := fmt.Sprintf("%s/r/talk/m/%s", OBSBaseURL, oid) - req, err := http.NewRequest("POST", url, bytes.NewReader(data)) + req, err := http.NewRequestWithContext(c.requestContext(), "POST", url, bytes.NewReader(data)) if err != nil { return fmt.Errorf("failed to create OBS request: %w", err) } @@ -664,7 +664,7 @@ func (c *Client) UploadOBSPlain(data []byte, oid string, obsType string) error { body, _ := io.ReadAll(resp.Body) if resp.StatusCode != 200 && resp.StatusCode != 201 { - return fmt.Errorf("OBS upload failed (%d): %s", resp.StatusCode, string(body)) + return newHTTPError(resp.StatusCode, "OBS upload failed (%d): %s", resp.StatusCode, string(body)) } return nil @@ -693,7 +693,7 @@ func (c *Client) UploadOBSWithSID(data []byte, sid string) (string, error) { url := fmt.Sprintf("%s/r/talk/%s/reqid-%s", OBSBaseURL, sid, reqID) - req, err := http.NewRequest("POST", url, bytes.NewReader(data)) + req, err := http.NewRequestWithContext(c.requestContext(), "POST", url, bytes.NewReader(data)) if err != nil { return "", fmt.Errorf("failed to create OBS request: %w", err) } @@ -726,7 +726,7 @@ func (c *Client) UploadOBSWithSID(data []byte, sid string) (string, error) { body, _ := io.ReadAll(resp.Body) if resp.StatusCode != 200 && resp.StatusCode != 201 { - return "", fmt.Errorf("OBS upload failed (%d): %s", resp.StatusCode, string(body)) + return "", newHTTPError(resp.StatusCode, "OBS upload failed (%d): %s", resp.StatusCode, string(body)) } // The OID is typically returned in the x-obs-oid header @@ -754,7 +754,7 @@ func (c *Client) UploadOBSWithOIDAndSID(data []byte, oid string, sid string) err url := fmt.Sprintf("%s/r/talk/%s/%s", OBSBaseURL, sid, oid) - req, err := http.NewRequest("POST", url, bytes.NewReader(data)) + req, err := http.NewRequestWithContext(c.requestContext(), "POST", url, bytes.NewReader(data)) if err != nil { return fmt.Errorf("failed to create OBS request: %w", err) } @@ -784,7 +784,7 @@ func (c *Client) UploadOBSWithOIDAndSID(data []byte, oid string, sid string) err body, _ := io.ReadAll(resp.Body) if resp.StatusCode != 200 && resp.StatusCode != 201 { - return fmt.Errorf("OBS upload failed (%d): %s", resp.StatusCode, string(body)) + return newHTTPError(resp.StatusCode, "OBS upload failed (%d): %s", resp.StatusCode, string(body)) } return nil @@ -842,7 +842,7 @@ func (c *Client) DownloadOBSPublicResource(ctx context.Context, resourcePath str case http.StatusNotFound: return nil, ErrOBSObjectNotFound default: - return nil, fmt.Errorf("public OBS resource download failed (%d): %s", resp.StatusCode, string(body)) + return nil, newHTTPError(resp.StatusCode, "public OBS resource download failed (%d): %s", resp.StatusCode, string(body)) } } @@ -902,7 +902,7 @@ func (c *Client) DownloadAlbumPreview(ctx context.Context, oid, chatID, albumID case http.StatusNotFound: return nil, ErrOBSObjectNotFound default: - return nil, fmt.Errorf("OBS download failed (%d): %s", resp.StatusCode, string(body)) + return nil, newHTTPError(resp.StatusCode, "OBS download failed (%d): %s", resp.StatusCode, string(body)) } } @@ -986,7 +986,7 @@ func (c *Client) checkOBSObjectReady(ctx context.Context, obsURL, obsToken, mess return fmt.Errorf("failed to read OBS object info response: %w", readErr) } if resp.StatusCode != http.StatusOK { - return fmt.Errorf("OBS object info failed (%d): %s", resp.StatusCode, string(body)) + return newHTTPError(resp.StatusCode, "OBS object info failed (%d): %s", resp.StatusCode, string(body)) } var info obsObjectInfo @@ -1026,7 +1026,7 @@ func (c *Client) downloadOBSObject(ctx context.Context, obsURL, obsToken, messag case http.StatusAccepted: return nil, ErrOBSEncodingIncomplete default: - return nil, fmt.Errorf("OBS download failed (%d): %s", resp.StatusCode, string(body)) + return nil, newHTTPError(resp.StatusCode, "OBS download failed (%d): %s", resp.StatusCode, string(body)) } } @@ -1090,7 +1090,7 @@ func (c *Client) constructTalkMeta(messageID string) string { func (c *Client) GetPageInfo(url string) (*PageInfoResult, error) { apiURL := "https://legy-jp.line-apps.com/sc/api/v2/pageinfo/get" - req, err := http.NewRequest("GET", apiURL, nil) + req, err := http.NewRequestWithContext(c.requestContext(), "GET", apiURL, nil) if err != nil { return nil, err } @@ -1113,7 +1113,7 @@ func (c *Client) GetPageInfo(url string) (*PageInfoResult, error) { defer resp.Body.Close() if resp.StatusCode != 200 { - return nil, fmt.Errorf("pageinfo request failed: %d", resp.StatusCode) + return nil, newHTTPError(resp.StatusCode, "pageinfo request failed: %d", resp.StatusCode) } var wrapper PageInfoResponse diff --git a/pkg/line/errors.go b/pkg/line/errors.go index dae5c2cb..6a80d976 100644 --- a/pkg/line/errors.go +++ b/pkg/line/errors.go @@ -18,6 +18,40 @@ var ( ErrGroupKeyNotFound = errors.New("group key not found") ) +// HTTPError is returned when a LINE endpoint answers with an unexpected HTTP status. +type HTTPError struct { + StatusCode int + text string +} + +func newHTTPError(statusCode int, format string, args ...any) *HTTPError { + return &HTTPError{StatusCode: statusCode, text: fmt.Sprintf(format, args...)} +} + +func (e *HTTPError) Error() string { + return e.text +} + +// HTTPStatusCode returns the HTTP status of a failed LINE request, or 0 if the +// request didn't get an HTTP response. +func HTTPStatusCode(err error) int { + var httpErr *HTTPError + if errors.As(err, &httpErr) { + return httpErr.StatusCode + } + return 0 +} + +// IsTalkException returns true when LINE rejected a Talk API call with a +// TalkException response. +func IsTalkException(err error) bool { + if err == nil { + return false + } + msg := strings.ToLower(err.Error()) + return hasResponseErrorCode(msg) && strings.Contains(msg, "talkexception") +} + type tokenRefreshError struct { code int loggedOut bool diff --git a/pkg/line/sse.go b/pkg/line/sse.go index faf839d2..63a30798 100644 --- a/pkg/line/sse.go +++ b/pkg/line/sse.go @@ -20,6 +20,10 @@ var sseHTTPClient = &http.Client{} var ErrSSEIdleTimeout = errors.New("SSE idle timeout") +// SSEEventOpen is passed to the ListenSSE callback with empty data once the +// server has accepted the stream, before any server-sent events. +const SSEEventOpen = "open" + const maxSSEErrorBodyBytes = 4096 // LINE advertises a 140s polling timeout. The extra 10s lets a normal @@ -65,10 +69,11 @@ func (c *Client) ListenSSE(ctx context.Context, localRev int64, callback func(ev if resp.StatusCode != 200 { body, _ := io.ReadAll(io.LimitReader(resp.Body, maxSSEErrorBodyBytes)) if bodyText := strings.TrimSpace(string(body)); bodyText != "" { - return fmt.Errorf("SSE error: %d: %s", resp.StatusCode, bodyText) + return newHTTPError(resp.StatusCode, "SSE error: %d: %s", resp.StatusCode, bodyText) } - return fmt.Errorf("SSE error: %d", resp.StatusCode) + return newHTTPError(resp.StatusCode, "SSE error: %d", resp.StatusCode) } + callback(SSEEventOpen, "") reader := bufio.NewReader(resp.Body) idleTimeout := sseIdleTimeout