Method ListActiveSessions

repo/content/sessions.go:189–238

ListActiveSessions returns a set of all active sessions in a given storage.

(ctx context.Context)

// ListActiveSessions returns a set of all active sessions in a given storage.
func (bm *WriteManager) ListActiveSessions(ctx context.Context) (map[SessionID]*SessionInfo, error) {
	blobs, err := blob.ListAllBlobs(ctx, bm.st, BlobIDPrefixSession)
	if err != nil {
		return nil, errors.Wrap(err, "unable to list session blobs")
	}

	m := map[SessionID]*SessionInfo{}

	var payload gather.WriteBuffer
	defer payload.Close()

	var decrypted gather.WriteBuffer
	defer decrypted.Close()

	for _, b := range blobs {
		payload.Reset()
		decrypted.Reset()

		sid := SessionIDFromBlobID(b.BlobID)
		if sid == "" {
			return nil, errors.Errorf("found invalid session blob %v", b.BlobID)
		}

		si := &SessionInfo{}

		err := bm.st.GetBlob(ctx, b.BlobID, 0, -1, &payload)
		if err != nil {
			if errors.Is(err, blob.ErrBlobNotFound) {
				continue
			}

			return nil, errors.Wrapf(err, "error loading session: %v", b.BlobID)
		}

		err = blobcrypto.Decrypt(bm.format, payload.Bytes(), b.BlobID, &decrypted)
		if err != nil {
			return nil, errors.Wrapf(err, "error decrypting session: %v", b.BlobID)
		}

		if err := json.NewDecoder(decrypted.Bytes().Reader()).Decode(si); err != nil {
			return nil, errors.Wrapf(err, "error parsing session: %v", b.BlobID)
		}

		if old := m[sid]; old == nil || si.CheckpointTime.After(old.CheckpointTime) {
			m[sid] = si
		}
	}

	return m, nil
}