feat: implement first batch im enhancements

2026-08-21 11:42:45 +08:00
parent 14abe58b88
commit c89c466387
16 changed files with 2629 additions and 130 deletions
@@ -398,6 +398,97 @@ func (a *adapter) TenantGroupInvitationCreate(id t.TenantID, inv *t.GroupInvitat
 	return inv, nil
 }

+func (a *adapter) TenantGroupInvitationAccept(id t.TenantID, invitationID string, uid t.Uid, memberPrivate any, now time.Time) (*t.GroupInvitation, error) {
+	if tenantAdp, err := a.routeGroupTenant(id); err != nil {
+		return nil, err
+	} else if tenantAdp != a {
+		return tenantAdp.TenantGroupInvitationAccept(id, invitationID, uid, memberPrivate, now)
+	}
+	if strings.TrimSpace(invitationID) == "" || uid.IsZero() || now.IsZero() {
+		return nil, t.ErrMalformed
+	}
+	ctx, cancel := a.getContextForTx()
+	if cancel != nil {
+		defer cancel()
+	}
+	tx, err := a.db.BeginTxx(ctx, nil)
+	if err != nil {
+		return nil, err
+	}
+	defer tx.Rollback()
+
+	var inv t.GroupInvitation
+	var handledAt sql.NullTime
+	err = tx.QueryRowxContext(ctx, `SELECT id,invitation_id,topic,inviter_user_id,target_user_id,message,status,result,
+			join_request_id,idempotency_key,expires_at,handled_at,created_at,updated_at
+		FROM im_group_invitations WHERE invitation_id=? FOR UPDATE`, strings.TrimSpace(invitationID)).
+		Scan(&inv.ID, &inv.InvitationID, &inv.Topic, &inv.InviterUserRaw, &inv.TargetUserRaw, &inv.Message, &inv.Status,
+			&inv.Result, &inv.JoinRequestID, &inv.IdempotencyKey, &inv.ExpiresAt, &handledAt, &inv.CreatedAt, &inv.UpdatedAt)
+	if err == sql.ErrNoRows {
+		return nil, t.ErrNotFound
+	}
+	if err != nil {
+		return nil, err
+	}
+	inv.InviterUserID = store.EncodeUid(inv.InviterUserRaw)
+	inv.TargetUserID = store.EncodeUid(inv.TargetUserRaw)
+	if handledAt.Valid {
+		inv.HandledAt = &handledAt.Time
+	}
+	if inv.TargetUserID != uid {
+		return nil, t.ErrPermissionDenied
+	}
+	if inv.ExpiresAt.Before(now) {
+		return nil, t.ErrExpired
+	}
+	if inv.Status == t.GroupInvitationStatusJoined || inv.Result == "joined" {
+		return &inv, nil
+	}
+	var deletedAt sql.NullTime
+	err = tx.QueryRowxContext(ctx, "SELECT DeletedAt FROM subscriptions WHERE Topic=? AND User=?", inv.Topic, uid.String()).Scan(&deletedAt)
+	if err != nil && err != sql.ErrNoRows {
+		return nil, err
+	}
+	if err == sql.ErrNoRows || deletedAt.Valid {
+		sub := &t.Subscription{TenantID: id, User: uid.String(), Topic: inv.Topic,
+			ModeWant: t.ModeCPublic, ModeGiven: t.ModeCPublic, Private: memberPrivate}
+		sub.InitTimes()
+		if err = createSubscription(tx, sub, deletedAt.Valid); err != nil {
+			return nil, err
+		}
+	}
+	member := &t.GroupMember{Topic: inv.Topic, UserID: uid, Role: t.GroupMemberRoleMember,
+		MemberStatus: t.GroupMemberStatusActive, JoinSource: "invitation", JoinedBy: inv.InviterUserID, JoinedAt: now,
+		Version: 1, CreatedAt: now, UpdatedAt: now}
+	if err = upsertGroupMemberTx(tx, member); err != nil {
+		return nil, err
+	}
+	if err = recountGroupMembersTx(tx, inv.Topic); err != nil {
+		return nil, err
+	}
+	if err = updateGroupTopicSubCountTx(tx, inv.Topic); err != nil {
+		return nil, err
+	}
+	if _, err = tx.ExecContext(ctx, `UPDATE im_group_invitations
+		SET status=?,result=?,handled_at=?,updated_at=? WHERE invitation_id=?`,
+		t.GroupInvitationStatusJoined, "joined", now, now, inv.InvitationID); err != nil {
+		return nil, err
+	}
+	if err = insertGroupEventTx(tx, inv.Topic, uid, "group.member.joined", map[string]any{
+		"user_id": uid.UserId(), "source": "invitation", "invitation_id": inv.InvitationID,
+	}); err != nil {
+		return nil, err
+	}
+	if err = tx.Commit(); err != nil {
+		return nil, err
+	}
+	inv.Status = t.GroupInvitationStatusJoined
+	inv.Result = "joined"
+	inv.HandledAt = &now
+	inv.UpdatedAt = now
+	return &inv, nil
+}
+
 func (a *adapter) TenantGroupJoinRequestCreate(id t.TenantID, req *t.GroupJoinRequest) (*t.GroupJoinRequest, error) {
 	if tenantAdp, err := a.routeGroupTenant(id); err != nil {
 		return nil, err
@@ -765,6 +856,130 @@ func (a *adapter) TenantGroupMuteDelete(id t.TenantID, topic string, uid, actor
 	return tx.Commit()
 }

+func (a *adapter) TenantGroupFileList(id t.TenantID, topic, fileType, query, cursor string, limit int) ([]t.GroupFile, string, error) {
+	if tenantAdp, err := a.routeGroupTenant(id); err != nil {
+		return nil, "", err
+	} else if tenantAdp != a {
+		return tenantAdp.TenantGroupFileList(id, topic, fileType, query, cursor, limit)
+	}
+	if strings.TrimSpace(topic) == "" {
+		return nil, "", t.ErrMalformed
+	}
+	where := []string{"topic=?", "status='active'"}
+	args := []any{strings.TrimSpace(topic)}
+	if fileType != "" {
+		where = append(where, "file_type=?")
+		args = append(args, strings.TrimSpace(fileType))
+	}
+	if query != "" {
+		where = append(where, "file_name LIKE ?")
+		args = append(args, "%"+strings.TrimSpace(query)+"%")
+	}
+	if cursor != "" {
+		parsed, err := strconv.ParseInt(cursor, 10, 64)
+		if err != nil || parsed <= 0 {
+			return nil, "", t.ErrMalformed
+		}
+		where = append(where, "id<?")
+		args = append(args, parsed)
+	}
+	args = append(args, limit+1)
+	ctx, cancel := a.getContext()
+	if cancel != nil {
+		defer cancel()
+	}
+	rows, err := a.db.QueryxContext(ctx, `SELECT id,group_file_id,topic,file_id,source_topic,source_seq,uploader_id,
+			file_name,mime_type,file_type,file_size,status,deleted_by,deleted_at,idempotency_key,created_at,updated_at
+		FROM im_group_files WHERE `+strings.Join(where, " AND ")+` ORDER BY id DESC LIMIT ?`, args...)
+	if err != nil {
+		return nil, "", err
+	}
+	defer rows.Close()
+	items := make([]t.GroupFile, 0, limit)
+	next := ""
+	for rows.Next() {
+		file, err := scanGroupFile(rows)
+		if err != nil {
+			return nil, "", err
+		}
+		if len(items) == limit {
+			next = strconv.FormatInt(file.ID, 10)
+			continue
+		}
+		items = append(items, *file)
+	}
+	return items, next, rows.Err()
+}
+
+func (a *adapter) TenantGroupFileCreate(id t.TenantID, file *t.GroupFile) (*t.GroupFile, error) {
+	if tenantAdp, err := a.routeGroupTenant(id); err != nil {
+		return nil, err
+	} else if tenantAdp != a {
+		return tenantAdp.TenantGroupFileCreate(id, file)
+	}
+	if file == nil || strings.TrimSpace(file.Topic) == "" || strings.TrimSpace(file.FileID) == "" || file.UploaderID.IsZero() {
+		return nil, t.ErrMalformed
+	}
+	if file.GroupFileID == "" {
+		file.GroupFileID = "gfile_" + store.Store.GetUidString()
+	}
+	if file.FileType == "" {
+		file.FileType = "file"
+	}
+	if file.Status == "" {
+		file.Status = "active"
+	}
+	if file.IdempotencyKey == "" {
+		file.IdempotencyKey = file.GroupFileID
+	}
+	now := t.TimeNow()
+	ctx, cancel := a.getContext()
+	if cancel != nil {
+		defer cancel()
+	}
+	_, err := a.db.ExecContext(ctx, `INSERT INTO im_group_files
+		(group_file_id,topic,file_id,source_topic,source_seq,uploader_id,file_name,mime_type,file_type,file_size,status,idempotency_key,created_at,updated_at)
+		VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?)
+		ON DUPLICATE KEY UPDATE updated_at=updated_at`,
+		file.GroupFileID, file.Topic, file.FileID, file.SourceTopic, file.SourceSeq, decodeUID(file.UploaderID),
+		file.FileName, file.MIMEType, file.FileType, file.FileSize, file.Status, file.IdempotencyKey, now, now)
+	if err != nil && !isDupe(err) {
+		return nil, err
+	}
+	return a.groupFileByQuery(ctx, "SELECT * FROM im_group_files WHERE topic=? AND file_id=?", file.Topic, file.FileID)
+}
+
+func (a *adapter) TenantGroupFileDelete(id t.TenantID, topic, groupFileID string, actor t.Uid, canManage bool, now time.Time) (*t.GroupFile, error) {
+	if tenantAdp, err := a.routeGroupTenant(id); err != nil {
+		return nil, err
+	} else if tenantAdp != a {
+		return tenantAdp.TenantGroupFileDelete(id, topic, groupFileID, actor, canManage, now)
+	}
+	if strings.TrimSpace(topic) == "" || strings.TrimSpace(groupFileID) == "" || actor.IsZero() || now.IsZero() {
+		return nil, t.ErrMalformed
+	}
+	ctx, cancel := a.getContext()
+	if cancel != nil {
+		defer cancel()
+	}
+	query := `UPDATE im_group_files
+		SET status='deleted',deleted_by=?,deleted_at=?,updated_at=?
+		WHERE topic=? AND group_file_id=? AND status='active'`
+	args := []any{decodeUID(actor), now, now, strings.TrimSpace(topic), strings.TrimSpace(groupFileID)}
+	if !canManage {
+		query += " AND uploader_id=?"
+		args = append(args, decodeUID(actor))
+	}
+	res, err := a.db.ExecContext(ctx, query, args...)
+	if err != nil {
+		return nil, err
+	}
+	if affected, _ := res.RowsAffected(); affected == 0 {
+		return nil, t.ErrNotFound
+	}
+	return a.groupFileByQuery(ctx, "SELECT * FROM im_group_files WHERE topic=? AND group_file_id=?", strings.TrimSpace(topic), strings.TrimSpace(groupFileID))
+}
+
 func (a *adapter) TenantGroupCreate(id t.TenantID, topic *t.Topic, owner t.Uid, ownerPrivate any,
 	settings *t.GroupSettings, members []t.Uid, memberPrivate any, now time.Time) error {
 	if tenantAdp, err := a.routeGroupTenant(id); err != nil {
@@ -1230,6 +1445,21 @@ func (a *adapter) groupMuteByQuery(ctx context.Context, query string, args ...an
 	return &item, nil
 }

+func (a *adapter) groupFileByQuery(ctx context.Context, query string, args ...any) (*t.GroupFile, error) {
+	rows, err := a.db.QueryxContext(ctx, query, args...)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	if rows.Next() {
+		return scanGroupFile(rows)
+	}
+	if err = rows.Err(); err != nil {
+		return nil, err
+	}
+	return nil, t.ErrNotFound
+}
+
 func (a *adapter) recountGroupMembers(ctx context.Context, topic string) error {
 	_, err := a.db.ExecContext(ctx, "UPDATE im_group_settings SET member_count=(SELECT COUNT(*) FROM im_group_members WHERE topic=? AND member_status=?),updated_at=CURRENT_TIMESTAMP(3) WHERE topic=?", topic, t.GroupMemberStatusActive, topic)
 	return err
@@ -1270,3 +1500,15 @@ func scanGroupMute(item *t.GroupMute) {
 		item.RevokedBy = store.EncodeUid(*item.RevokedRaw)
 	}
 }
+
+func scanGroupFile(rows *sqlx.Rows) (*t.GroupFile, error) {
+	var item t.GroupFile
+	if err := rows.StructScan(&item); err != nil {
+		return nil, err
+	}
+	item.UploaderID = store.EncodeUid(item.UploaderRaw)
+	if item.DeletedByRaw != nil {
+		item.DeletedBy = store.EncodeUid(*item.DeletedByRaw)
+	}
+	return &item, nil
+}