From 566524d0a9d086f253f7ad28a91e8986b8400066 Mon Sep 17 00:00:00 2001
From: trilero <trile@riseup.net>
Date: Thu, 18 Apr 2024 08:33:21 +0200
Subject: [PATCH] [+][FCB] Compute numbers of soft copies after pinning.
---
.../acab/devcon0/boot/ioc/FederationIoC.scala | 3 +-
.../scala/acab/devcon0/boot/ioc/IpfsIoC.scala | 27 +++++--
.../HeartbeatCommandHandlerImpl.scala | 2 +-
...IpfsCidDeltaUpdateCommandHandlerImpl.scala | 34 ++++++--
.../CheckInAckCommandHandlerImpl.scala | 2 +-
.../UpdateInformationCommandHandlerImpl.scala | 2 +-
.../ipfscid/MetadataQueryHandlerImpl.scala | 19 +++--
.../ipfscid/SearchQueryHandlerImpl.scala | 13 ++-
.../output/client/IpfsClusterClient.scala | 12 +--
...itory.scala => HardCopiesRepository.scala} | 2 +-
.../ipfscid/SoftCopiesRepository.scala | 11 +++
.../domain/service/IpfsClusterService.scala | 15 ++--
.../service/IpfsClusterServiceImpl.scala | 7 +-
.../domain/service/IpfsServiceImpl.scala | 3 +-
...sService.scala => HardCopiesService.scala} | 2 +-
...Impl.scala => HardCopiesServiceImpl.scala} | 4 +-
.../service/ipfscid/SoftCopiesService.scala | 9 +++
.../ipfscid/SoftCopiesServiceImpl.scala | 20 +++++
...ionMemberSharingFolderUpdateListener.scala | 3 +-
.../output/client/IpfsClusterClientImpl.scala | 12 ++-
...yImpl.scala => HardCopiesRepository.scala} | 12 +--
...yImpl.scala => SearchNameRepository.scala} | 0
.../ipfscid/SoftCopiesRepository.scala | 80 +++++++++++++++++++
.../output/repository/redisutils/Redis.scala | 11 +--
24 files changed, 235 insertions(+), 70 deletions(-)
rename federation-controller-backend/src/main/scala/acab/devcon0/domain/ports/output/repository/ipfscid/{CopiesRepository.scala => HardCopiesRepository.scala} (87%)
create mode 100644 federation-controller-backend/src/main/scala/acab/devcon0/domain/ports/output/repository/ipfscid/SoftCopiesRepository.scala
rename federation-controller-backend/src/main/scala/acab/devcon0/domain/service/ipfscid/{CopiesService.scala => HardCopiesService.scala} (88%)
rename federation-controller-backend/src/main/scala/acab/devcon0/domain/service/ipfscid/{CopiesServiceImpl.scala => HardCopiesServiceImpl.scala} (75%)
create mode 100644 federation-controller-backend/src/main/scala/acab/devcon0/domain/service/ipfscid/SoftCopiesService.scala
create mode 100644 federation-controller-backend/src/main/scala/acab/devcon0/domain/service/ipfscid/SoftCopiesServiceImpl.scala
rename federation-controller-backend/src/main/scala/acab/devcon0/output/repository/ipfscid/{CopiesRepositoryImpl.scala => HardCopiesRepository.scala} (86%)
rename federation-controller-backend/src/main/scala/acab/devcon0/output/repository/ipfscid/{SearchNameRepositoryImpl.scala => SearchNameRepository.scala} (100%)
create mode 100644 federation-controller-backend/src/main/scala/acab/devcon0/output/repository/ipfscid/SoftCopiesRepository.scala
diff --git a/federation-controller-backend/src/main/scala/acab/devcon0/boot/ioc/FederationIoC.scala b/federation-controller-backend/src/main/scala/acab/devcon0/boot/ioc/FederationIoC.scala
index e8cf665..ebfcd7d 100644
--- a/federation-controller-backend/src/main/scala/acab/devcon0/boot/ioc/FederationIoC.scala
+++ b/federation-controller-backend/src/main/scala/acab/devcon0/boot/ioc/FederationIoC.scala
@@ -45,7 +45,8 @@ object FederationIoC {
val ipfsCidUpdateCommandHandler: IpfsCidDeltaUpdateCommandHandler = {
new IpfsCidDeltaUpdateCommandHandlerImpl(
ipfsCidFacadeService = IpfsIoC.Domain.Service.facadeService,
- ipfsCidCopiesService = IpfsIoC.Domain.Service.copiesService,
+ ipfsCidHardCopiesService = IpfsIoC.Domain.Service.hardCopiesService,
+ ipfsCidSoftCopiesService = IpfsIoC.Domain.Service.softCopiesService,
ipfsClusterService = IpfsIoC.Domain.Service.ipfsClusterService
)
}
diff --git a/federation-controller-backend/src/main/scala/acab/devcon0/boot/ioc/IpfsIoC.scala b/federation-controller-backend/src/main/scala/acab/devcon0/boot/ioc/IpfsIoC.scala
index 6cb37e8..f48aa8a 100644
--- a/federation-controller-backend/src/main/scala/acab/devcon0/boot/ioc/IpfsIoC.scala
+++ b/federation-controller-backend/src/main/scala/acab/devcon0/boot/ioc/IpfsIoC.scala
@@ -6,8 +6,9 @@ import acab.devcon0.domain.adapters.ipfscid._
import acab.devcon0.domain.ports.input.ipfscid._
import acab.devcon0.domain.ports.output.client.IpfsClient
import acab.devcon0.domain.ports.output.client.IpfsClusterClient
-import acab.devcon0.domain.ports.output.repository.ipfscid.CopiesRepository
+import acab.devcon0.domain.ports.output.repository.ipfscid.HardCopiesRepository
import acab.devcon0.domain.ports.output.repository.ipfscid.SearchNameRepository
+import acab.devcon0.domain.ports.output.repository.ipfscid.SoftCopiesRepository
import acab.devcon0.domain.service.IpfsClusterService
import acab.devcon0.domain.service.IpfsClusterServiceImpl
import acab.devcon0.domain.service.IpfsService
@@ -45,13 +46,15 @@ object IpfsIoC {
val metaDataQueryHandler: MetadataQueryHandler = MetadataQueryHandlerImpl(
facadeService = Service.facadeService,
ipfsService = Domain.Service.ipfsService,
- copiesService = Service.copiesService
+ hardCopiesService = Service.hardCopiesService,
+ softCopiesService = Service.softCopiesService
)
val searchQueryHandler: SearchQueryHandler = SearchQueryHandlerImpl(
facadeService = Service.facadeService,
searchNameService = Service.searchNameService,
- copiesService = Service.copiesService
+ hardCopiesService = Service.hardCopiesService,
+ softCopiesService = Service.softCopiesService
)
}
@@ -62,7 +65,7 @@ object IpfsIoC {
flatRepository = Output.flatRepository
)
- val ipfsClusterService: IpfsClusterService = new IpfsClusterServiceImpl(
+ val ipfsClusterService: IpfsClusterService[IO] = new IpfsClusterServiceImpl(
client = Output.ipfsClusterClient
)
@@ -70,8 +73,12 @@ object IpfsIoC {
repository = Output.searchNameRepository
)
- val copiesService: CopiesService[IO] = new CopiesServiceImpl(
- repository = Output.copiesRepository
+ val hardCopiesService: HardCopiesService[IO] = new HardCopiesServiceImpl(
+ repository = Output.hardCopiesRepository
+ )
+
+ val softCopiesService: SoftCopiesService[IO] = new SoftCopiesServiceImpl(
+ repository = Output.softCopiesRepository
)
val facadeService: FacadeService[IO] = new FacadeServiceImpl(
@@ -87,7 +94,7 @@ object IpfsIoC {
private val backend: SyncBackend = HttpURLConnectionBackend()
private val backendResource: Resource[IO, SyncBackend] = Resource.make(IO(backend))(res => IO(res.close()))
- val ipfsClusterClient: IpfsClusterClient = new IpfsClusterClientImpl(
+ val ipfsClusterClient: IpfsClusterClient[IO] = new IpfsClusterClientImpl(
apiUrl = Configuration.configuration.ipfsCluster.apiUrl,
backendResource = backendResource
)
@@ -97,7 +104,11 @@ object IpfsIoC {
ipfsConfiguration = Configuration.configuration.ipfs
)
- val copiesRepository: CopiesRepository[IO] = new CopiesRepositoryImpl(
+ val hardCopiesRepository: HardCopiesRepository[IO] = new HardCopiesRepositoryImpl(
+ commandsApi = CommonsIoC.InputOutput.redisCommandsFactory()
+ )
+
+ val softCopiesRepository: SoftCopiesRepository[IO] = new SoftCopiesRepositoryImpl(
commandsApi = CommonsIoC.InputOutput.redisCommandsFactory()
)
diff --git a/federation-controller-backend/src/main/scala/acab/devcon0/domain/adapters/federation/HeartbeatCommandHandlerImpl.scala b/federation-controller-backend/src/main/scala/acab/devcon0/domain/adapters/federation/HeartbeatCommandHandlerImpl.scala
index b90df0f..2d1b958 100644
--- a/federation-controller-backend/src/main/scala/acab/devcon0/domain/adapters/federation/HeartbeatCommandHandlerImpl.scala
+++ b/federation-controller-backend/src/main/scala/acab/devcon0/domain/adapters/federation/HeartbeatCommandHandlerImpl.scala
@@ -9,7 +9,7 @@ import cats.effect.IO
import org.typelevel.log4cats.Logger
import org.typelevel.log4cats.slf4j.Slf4jLogger
-class HeartbeatCommandHandlerImpl(p2PService: P2pService[IO], ipfsClusterService: IpfsClusterService)
+class HeartbeatCommandHandlerImpl(p2PService: P2pService[IO], ipfsClusterService: IpfsClusterService[IO])
extends HeartbeatCommandHandler {
implicit val logger: Logger[IO] = Slf4jLogger.getLogger[IO]
diff --git a/federation-controller-backend/src/main/scala/acab/devcon0/domain/adapters/federation/IpfsCidDeltaUpdateCommandHandlerImpl.scala b/federation-controller-backend/src/main/scala/acab/devcon0/domain/adapters/federation/IpfsCidDeltaUpdateCommandHandlerImpl.scala
index f55d9a2..b6c7737 100644
--- a/federation-controller-backend/src/main/scala/acab/devcon0/domain/adapters/federation/IpfsCidDeltaUpdateCommandHandlerImpl.scala
+++ b/federation-controller-backend/src/main/scala/acab/devcon0/domain/adapters/federation/IpfsCidDeltaUpdateCommandHandlerImpl.scala
@@ -1,18 +1,23 @@
package acab.devcon0.domain.adapters.federation
-import acab.devcon0.domain.ports.input._
+import scala.concurrent.duration.DurationInt
+
import acab.devcon0.domain.ports.input.federation.IpfsCidDeltaUpdateCommand
import acab.devcon0.domain.ports.input.federation.IpfsCidDeltaUpdateCommandHandler
import acab.devcon0.domain.ports.input.federation.IpfsCidDeltaUpdateEvent
import acab.devcon0.domain.ports.input.federation.IpfsCidDeltaUpdateSuccessEvent
import acab.devcon0.domain.service._
-import acab.devcon0.domain.service.ipfscid.CopiesService
import acab.devcon0.domain.service.ipfscid.FacadeService
+import acab.devcon0.domain.service.ipfscid.HardCopiesService
+import acab.devcon0.domain.service.ipfscid.SoftCopiesService
import cats.effect.IO
+import cats.effect.std.Supervisor
+import cats.implicits.toTraverseOps
class IpfsCidDeltaUpdateCommandHandlerImpl(
ipfsCidFacadeService: FacadeService[IO],
- ipfsCidCopiesService: CopiesService[IO],
- ipfsClusterService: IpfsClusterService
+ ipfsCidHardCopiesService: HardCopiesService[IO],
+ ipfsCidSoftCopiesService: SoftCopiesService[IO],
+ ipfsClusterService: IpfsClusterService[IO]
) extends IpfsCidDeltaUpdateCommandHandler {
override def handle(cmd: IpfsCidDeltaUpdateCommand): IO[IpfsCidDeltaUpdateEvent[?]] = {
@@ -24,16 +29,29 @@ class IpfsCidDeltaUpdateCommandHandlerImpl(
private def handleAdditions(cmd: IpfsCidDeltaUpdateCommand): IO[Unit] = {
for
- _ <- ipfsCidCopiesService.increment(cmd.delta.additions)
+ _ <- ipfsCidHardCopiesService.increment(cmd.delta.additions)
_ <- ipfsClusterService.addPins(cmd.delta.additions)
+ _ <- handleAdditionsSoftCopies(cmd)
yield ()
}
private def handleRemovals(cmd: IpfsCidDeltaUpdateCommand): IO[Unit] = {
for
- _ <- ipfsCidFacadeService.deleteFlats(cmd.delta.removals)
- _ <- ipfsCidCopiesService.decrement(cmd.delta.removals)
- _ <- ipfsClusterService.removePins(cmd.delta.removals)
+ _ <- ipfsCidFacadeService.deleteFlats(cmd.delta.removals)
+ _ <- ipfsCidHardCopiesService.decrement(cmd.delta.removals)
+ _ <- ipfsClusterService.removePins(cmd.delta.removals)
+ pinsAllocations <- cmd.delta.removals.toSeq.traverse(ipfsClusterService.getPinAllocations)
+ _ <- pinsAllocations.traverse(pin => ipfsCidSoftCopiesService.delete(pin.cid))
yield ()
}
+
+ private def handleAdditionsSoftCopies(cmd: IpfsCidDeltaUpdateCommand): IO[Unit] = {
+ Supervisor[IO](await = false).use(supervisor => {
+ for
+ _ <- IO.sleep(10.seconds)
+ pinsAllocations <- cmd.delta.additions.toSeq.traverse(ipfsClusterService.getPinAllocations)
+ _ <- pinsAllocations.traverse(pin => ipfsCidSoftCopiesService.set(pin.cid, pin.allocations.size))
+ yield ()
+ })
+ }
}
diff --git a/federation-controller-backend/src/main/scala/acab/devcon0/domain/adapters/federationmember/CheckInAckCommandHandlerImpl.scala b/federation-controller-backend/src/main/scala/acab/devcon0/domain/adapters/federationmember/CheckInAckCommandHandlerImpl.scala
index 91bd4f9..f1ab896 100644
--- a/federation-controller-backend/src/main/scala/acab/devcon0/domain/adapters/federationmember/CheckInAckCommandHandlerImpl.scala
+++ b/federation-controller-backend/src/main/scala/acab/devcon0/domain/adapters/federationmember/CheckInAckCommandHandlerImpl.scala
@@ -12,7 +12,7 @@ import cats.effect.IO
import org.typelevel.log4cats.Logger
import org.typelevel.log4cats.slf4j.Slf4jLogger
-class CheckInAckCommandHandlerImpl(p2PService: P2pService[IO], ipfsClusterService: IpfsClusterService)
+class CheckInAckCommandHandlerImpl(p2PService: P2pService[IO], ipfsClusterService: IpfsClusterService[IO])
extends CheckInAckCommandHandler {
implicit val logger: Logger[IO] = Slf4jLogger.getLogger[IO]
diff --git a/federation-controller-backend/src/main/scala/acab/devcon0/domain/adapters/federationmember/UpdateInformationCommandHandlerImpl.scala b/federation-controller-backend/src/main/scala/acab/devcon0/domain/adapters/federationmember/UpdateInformationCommandHandlerImpl.scala
index f0ada6e..268af22 100644
--- a/federation-controller-backend/src/main/scala/acab/devcon0/domain/adapters/federationmember/UpdateInformationCommandHandlerImpl.scala
+++ b/federation-controller-backend/src/main/scala/acab/devcon0/domain/adapters/federationmember/UpdateInformationCommandHandlerImpl.scala
@@ -17,7 +17,7 @@ import org.typelevel.log4cats.Logger
import org.typelevel.log4cats.slf4j.Slf4jLogger
class UpdateInformationCommandHandlerImpl(
- ipfsClusterService: IpfsClusterService,
+ ipfsClusterService: IpfsClusterService[IO],
informationService: InformationService[IO]
) extends UpdateInformationCommandHandler {
diff --git a/federation-controller-backend/src/main/scala/acab/devcon0/domain/adapters/ipfscid/MetadataQueryHandlerImpl.scala b/federation-controller-backend/src/main/scala/acab/devcon0/domain/adapters/ipfscid/MetadataQueryHandlerImpl.scala
index 5e400f2..db1003b 100644
--- a/federation-controller-backend/src/main/scala/acab/devcon0/domain/adapters/ipfscid/MetadataQueryHandlerImpl.scala
+++ b/federation-controller-backend/src/main/scala/acab/devcon0/domain/adapters/ipfscid/MetadataQueryHandlerImpl.scala
@@ -6,37 +6,40 @@ import acab.devcon0.domain.dtos.IpfsCidMetadataDto
import acab.devcon0.domain.mappers.IpfsCidMapper
import acab.devcon0.domain.ports.input.ipfscid._
import acab.devcon0.domain.service.IpfsService
-import acab.devcon0.domain.service.ipfscid.CopiesService
import acab.devcon0.domain.service.ipfscid.FacadeService
+import acab.devcon0.domain.service.ipfscid.HardCopiesService
+import acab.devcon0.domain.service.ipfscid.SoftCopiesService
import cats.effect.IO
+import cats.implicits.catsSyntaxTuple2Parallel
import org.typelevel.log4cats.Logger
import org.typelevel.log4cats.slf4j.Slf4jLogger
class MetadataQueryHandlerImpl(
ipfsService: IpfsService,
facadeService: FacadeService[IO],
- copiesService: CopiesService[IO]
+ hardCopiesService: HardCopiesService[IO],
+ softCopiesService: SoftCopiesService[IO]
) extends MetadataQueryHandler {
private val logger: Logger[IO] = Slf4jLogger.getLogger[IO]
override def handle(query: MetadataQuery): IO[Option[IpfsCidMetadataDto]] = {
for
- maybeCacheHit <- facadeService.getFlat(query.ipfsCid)
- hardCopies <- copiesService.get(query.ipfsCid)
+ maybeCacheHit <- facadeService.getFlat(query.ipfsCid)
+ (hardCopies, softCopies) <- (hardCopiesService.get(query.ipfsCid), softCopiesService.get(query.ipfsCid)).parTupled
result <- maybeCacheHit match
- case Some(value) => IO(Some(buildIpfsCidMetadataDto(value, hardCopies)))
+ case Some(value) => IO(Some(buildIpfsCidMetadataDto(value, hardCopies, softCopies)))
case None =>
ipfsService
.lsFileTree(query.ipfsCid)
.map(IpfsCidMapper.Flat.to)
.flatTap(facadeService.saveFlat)
- .map(dto => Some(buildIpfsCidMetadataDto(dto, hardCopies)))
+ .map(dto => Some(buildIpfsCidMetadataDto(dto, hardCopies, softCopies)))
.onError(logError)
yield result
}
- private def buildIpfsCidMetadataDto(dto: IpfsCidDto, hardCopies: Int): IpfsCidMetadataDto = {
+ private def buildIpfsCidMetadataDto(dto: IpfsCidDto, hardCopies: Int, softCopies: Int): IpfsCidMetadataDto = {
IpfsCidMetadataDto(
cid = dto.cid,
size = dto.size,
@@ -45,7 +48,7 @@ class MetadataQueryHandlerImpl(
contents = dto.contents,
timestamp = dto.timestamp,
`type` = dto.`type`,
- copies = IpfsCidMetadataCopiesDto(hard = hardCopies, soft = 0),
+ copies = IpfsCidMetadataCopiesDto(hard = hardCopies, soft = softCopies),
names = List()
)
}
diff --git a/federation-controller-backend/src/main/scala/acab/devcon0/domain/adapters/ipfscid/SearchQueryHandlerImpl.scala b/federation-controller-backend/src/main/scala/acab/devcon0/domain/adapters/ipfscid/SearchQueryHandlerImpl.scala
index 630d723..141a3c3 100644
--- a/federation-controller-backend/src/main/scala/acab/devcon0/domain/adapters/ipfscid/SearchQueryHandlerImpl.scala
+++ b/federation-controller-backend/src/main/scala/acab/devcon0/domain/adapters/ipfscid/SearchQueryHandlerImpl.scala
@@ -2,21 +2,20 @@ package acab.devcon0.domain.adapters.ipfscid
import acab.devcon0.domain.dtos._
import acab.devcon0.domain.ports.input.ipfscid._
-import acab.devcon0.domain.service.ipfscid.CopiesService
import acab.devcon0.domain.service.ipfscid.FacadeService
+import acab.devcon0.domain.service.ipfscid.HardCopiesService
import acab.devcon0.domain.service.ipfscid.SearchNameService
+import acab.devcon0.domain.service.ipfscid.SoftCopiesService
import cats.effect.IO
import cats.implicits._
-import org.typelevel.log4cats.slf4j.Slf4jLogger
class SearchQueryHandlerImpl(
facadeService: FacadeService[IO],
searchNameService: SearchNameService[IO],
- copiesService: CopiesService[IO]
+ hardCopiesService: HardCopiesService[IO],
+ softCopiesService: SoftCopiesService[IO]
) extends SearchQueryHandler {
- Slf4jLogger.getLogger[IO]
-
override def handle(query: SearchQuery): IO[IpfsCidSearchResponse] = {
for
searchResultIpfsCids <- searchNameService.search(query.searchParameters)
@@ -26,7 +25,7 @@ class SearchQueryHandlerImpl(
}
private def buildIpfsCidMetadataDto(dto: IpfsCidDto): IO[IpfsCidMetadataDto] = {
- for hardCopies <- copiesService.get(dto.cid)
+ for (hardCopies, softCopies) <- (hardCopiesService.get(dto.cid), softCopiesService.get(dto.cid)).parTupled
yield IpfsCidMetadataDto(
cid = dto.cid,
size = dto.size,
@@ -35,7 +34,7 @@ class SearchQueryHandlerImpl(
contents = dto.contents,
timestamp = dto.timestamp,
`type` = dto.`type`,
- copies = IpfsCidMetadataCopiesDto(hardCopies, 0),
+ copies = IpfsCidMetadataCopiesDto(hardCopies, softCopies),
names = List()
)
}
diff --git a/federation-controller-backend/src/main/scala/acab/devcon0/domain/ports/output/client/IpfsClusterClient.scala b/federation-controller-backend/src/main/scala/acab/devcon0/domain/ports/output/client/IpfsClusterClient.scala
index 00b9a62..521ce7e 100644
--- a/federation-controller-backend/src/main/scala/acab/devcon0/domain/ports/output/client/IpfsClusterClient.scala
+++ b/federation-controller-backend/src/main/scala/acab/devcon0/domain/ports/output/client/IpfsClusterClient.scala
@@ -2,11 +2,11 @@ package acab.devcon0.domain.ports.output.client
import acab.devcon0.trile.domain.dtos._
import acab.devcon0.trile.domain.dtos.aliases.IpfsCid
-import cats.effect.IO
-trait IpfsClusterClient {
- def getPeers: IO[IpfsClusterPeers]
- def getPeer: IO[IpfsClusterPeer]
- def removePin(pinCID: IpfsCid): IO[Unit]
- def addPin(cid: IpfsCid): IO[Unit]
+trait IpfsClusterClient[F[_]] {
+ def getPeers: F[IpfsClusterPeers]
+ def getPeer: F[IpfsClusterPeer]
+ def getPinAllocations(ipfsCid: IpfsCid): F[IpfsClusterPinAllocation]
+ def removePin(pinCID: IpfsCid): F[Unit]
+ def addPin(cid: IpfsCid): F[Unit]
}
diff --git a/federation-controller-backend/src/main/scala/acab/devcon0/domain/ports/output/repository/ipfscid/CopiesRepository.scala b/federation-controller-backend/src/main/scala/acab/devcon0/domain/ports/output/repository/ipfscid/HardCopiesRepository.scala
similarity index 87%
rename from federation-controller-backend/src/main/scala/acab/devcon0/domain/ports/output/repository/ipfscid/CopiesRepository.scala
rename to federation-controller-backend/src/main/scala/acab/devcon0/domain/ports/output/repository/ipfscid/HardCopiesRepository.scala
index 90b435c..7c14c49 100644
--- a/federation-controller-backend/src/main/scala/acab/devcon0/domain/ports/output/repository/ipfscid/CopiesRepository.scala
+++ b/federation-controller-backend/src/main/scala/acab/devcon0/domain/ports/output/repository/ipfscid/HardCopiesRepository.scala
@@ -1,7 +1,7 @@
package acab.devcon0.domain.ports.output.repository.ipfscid
import acab.devcon0.trile.domain.dtos.aliases.IpfsCid
-trait CopiesRepository[F[_]] {
+trait HardCopiesRepository[F[_]] {
def get(ipfsCid: IpfsCid): F[Int]
def increment(ipfsCids: Set[IpfsCid]): F[Unit]
def decrement(ipfsCids: Set[IpfsCid]): F[Unit]
diff --git a/federation-controller-backend/src/main/scala/acab/devcon0/domain/ports/output/repository/ipfscid/SoftCopiesRepository.scala b/federation-controller-backend/src/main/scala/acab/devcon0/domain/ports/output/repository/ipfscid/SoftCopiesRepository.scala
new file mode 100644
index 0000000..7eed9dc
--- /dev/null
+++ b/federation-controller-backend/src/main/scala/acab/devcon0/domain/ports/output/repository/ipfscid/SoftCopiesRepository.scala
@@ -0,0 +1,11 @@
+package acab.devcon0.domain.ports.output.repository.ipfscid
+
+import acab.devcon0.trile.domain.dtos.aliases.IpfsCid
+
+trait SoftCopiesRepository[F[_]] {
+ def get(ipfsCid: IpfsCid): F[Int]
+
+ def set(ipfsCid: IpfsCid, count: Int): F[Unit]
+
+ def delete(ipfsCid: IpfsCid): F[Unit]
+}
diff --git a/federation-controller-backend/src/main/scala/acab/devcon0/domain/service/IpfsClusterService.scala b/federation-controller-backend/src/main/scala/acab/devcon0/domain/service/IpfsClusterService.scala
index b0729ea..efd8b5a 100644
--- a/federation-controller-backend/src/main/scala/acab/devcon0/domain/service/IpfsClusterService.scala
+++ b/federation-controller-backend/src/main/scala/acab/devcon0/domain/service/IpfsClusterService.scala
@@ -2,14 +2,15 @@ package acab.devcon0.domain.service
import acab.devcon0.trile.domain.dtos.IpfsClusterPeer
import acab.devcon0.trile.domain.dtos.IpfsClusterPeers
+import acab.devcon0.trile.domain.dtos.IpfsClusterPinAllocation
import acab.devcon0.trile.domain.dtos.aliases.IpfsCid
import acab.devcon0.trile.domain.dtos.aliases.P2pPeerId
-import cats.effect._
-trait IpfsClusterService {
- def getPeers: IO[IpfsClusterPeers]
- def getPeer(p2pPeerId: P2pPeerId): IO[IpfsClusterPeer]
- def getPeer: IO[IpfsClusterPeer]
- def addPins(ipfsCids: Set[IpfsCid]): IO[Unit]
- def removePins(ipfsCids: Set[IpfsCid]): IO[Unit]
+trait IpfsClusterService[F[_]] {
+ def getPeers: F[IpfsClusterPeers]
+ def getPeer(p2pPeerId: P2pPeerId): F[IpfsClusterPeer]
+ def getPeer: F[IpfsClusterPeer]
+ def addPins(ipfsCids: Set[IpfsCid]): F[Unit]
+ def getPinAllocations(ipfsCid: IpfsCid): F[IpfsClusterPinAllocation]
+ def removePins(ipfsCids: Set[IpfsCid]): F[Unit]
}
diff --git a/federation-controller-backend/src/main/scala/acab/devcon0/domain/service/IpfsClusterServiceImpl.scala b/federation-controller-backend/src/main/scala/acab/devcon0/domain/service/IpfsClusterServiceImpl.scala
index a251f66..e527043 100644
--- a/federation-controller-backend/src/main/scala/acab/devcon0/domain/service/IpfsClusterServiceImpl.scala
+++ b/federation-controller-backend/src/main/scala/acab/devcon0/domain/service/IpfsClusterServiceImpl.scala
@@ -3,6 +3,7 @@ package acab.devcon0.domain.service
import acab.devcon0.domain.ports.output.client.IpfsClusterClient
import acab.devcon0.trile.domain.dtos.IpfsClusterPeer
import acab.devcon0.trile.domain.dtos.IpfsClusterPeers
+import acab.devcon0.trile.domain.dtos.IpfsClusterPinAllocation
import acab.devcon0.trile.domain.dtos.aliases.IpfsCid
import acab.devcon0.trile.domain.dtos.aliases.IpfsClusterPeerId
import acab.devcon0.trile.utils.EffectsUtils
@@ -12,7 +13,7 @@ import cats.implicits.toTraverseOps
import org.typelevel.log4cats.Logger
import org.typelevel.log4cats.slf4j.Slf4jLogger
-class IpfsClusterServiceImpl(client: IpfsClusterClient) extends IpfsClusterService {
+class IpfsClusterServiceImpl(client: IpfsClusterClient[IO]) extends IpfsClusterService[IO] {
implicit val logger: Logger[IO] = Slf4jLogger.getLogger[IO]
@@ -32,6 +33,10 @@ class IpfsClusterServiceImpl(client: IpfsClusterClient) extends IpfsClusterServi
IORetry.fibonacci(getPeerIO)
}
+ override def getPinAllocations(ipfsCid: IpfsCid): IO[IpfsClusterPinAllocation] = {
+ client.getPinAllocations(ipfsCid)
+ }
+
override def addPins(ipfsCids: Set[IpfsCid]): IO[Unit] = {
ipfsCids.toSeq.traverse(addPin).void
}
diff --git a/federation-controller-backend/src/main/scala/acab/devcon0/domain/service/IpfsServiceImpl.scala b/federation-controller-backend/src/main/scala/acab/devcon0/domain/service/IpfsServiceImpl.scala
index 0e4c459..bc790cd 100644
--- a/federation-controller-backend/src/main/scala/acab/devcon0/domain/service/IpfsServiceImpl.scala
+++ b/federation-controller-backend/src/main/scala/acab/devcon0/domain/service/IpfsServiceImpl.scala
@@ -83,7 +83,7 @@ class IpfsServiceImpl(ipfsClient: IpfsClient, flatRepository: FlatRepository) ex
case Some(value) => IO(value)
case None => ipfsClientLs(ipfsCid).flatMap(buildIpfsCidDto(_, None))
})
- case Right(link) => {
+ case Right(link) =>
flatRepository
.get(link.hash)
.flatMap(maybeCacheHit => {
@@ -91,7 +91,6 @@ class IpfsServiceImpl(ipfsClient: IpfsClient, flatRepository: FlatRepository) ex
case Some(value) => IO(value)
case None => ipfsClientLs(link.hash).flatMap(buildIpfsCidDto(_, Some(link)))
})
- }
}
private def getIpfsCidDtos(links: List[IpfsLsLinkResponse]): IO[List[IpfsCidDto]] = {
diff --git a/federation-controller-backend/src/main/scala/acab/devcon0/domain/service/ipfscid/CopiesService.scala b/federation-controller-backend/src/main/scala/acab/devcon0/domain/service/ipfscid/HardCopiesService.scala
similarity index 88%
rename from federation-controller-backend/src/main/scala/acab/devcon0/domain/service/ipfscid/CopiesService.scala
rename to federation-controller-backend/src/main/scala/acab/devcon0/domain/service/ipfscid/HardCopiesService.scala
index a4f9f38..9dbedaf 100644
--- a/federation-controller-backend/src/main/scala/acab/devcon0/domain/service/ipfscid/CopiesService.scala
+++ b/federation-controller-backend/src/main/scala/acab/devcon0/domain/service/ipfscid/HardCopiesService.scala
@@ -1,7 +1,7 @@
package acab.devcon0.domain.service.ipfscid
import acab.devcon0.trile.domain.dtos.aliases.IpfsCid
-trait CopiesService[F[_]] {
+trait HardCopiesService[F[_]] {
def get(ipfsCid: IpfsCid): F[Int]
def increment(ipfsCids: Set[IpfsCid]): F[Unit]
def decrement(ipfsCids: Set[IpfsCid]): F[Unit]
diff --git a/federation-controller-backend/src/main/scala/acab/devcon0/domain/service/ipfscid/CopiesServiceImpl.scala b/federation-controller-backend/src/main/scala/acab/devcon0/domain/service/ipfscid/HardCopiesServiceImpl.scala
similarity index 75%
rename from federation-controller-backend/src/main/scala/acab/devcon0/domain/service/ipfscid/CopiesServiceImpl.scala
rename to federation-controller-backend/src/main/scala/acab/devcon0/domain/service/ipfscid/HardCopiesServiceImpl.scala
index 6d9522c..9529333 100644
--- a/federation-controller-backend/src/main/scala/acab/devcon0/domain/service/ipfscid/CopiesServiceImpl.scala
+++ b/federation-controller-backend/src/main/scala/acab/devcon0/domain/service/ipfscid/HardCopiesServiceImpl.scala
@@ -1,11 +1,11 @@
package acab.devcon0.domain.service.ipfscid
-import acab.devcon0.domain.ports.output.repository.ipfscid.CopiesRepository
+import acab.devcon0.domain.ports.output.repository.ipfscid.HardCopiesRepository
import acab.devcon0.domain.service.ipfscid
import acab.devcon0.trile.domain.dtos.aliases.IpfsCid
import cats.effect.IO
-class CopiesServiceImpl(repository: CopiesRepository[IO]) extends CopiesService[IO] {
+class HardCopiesServiceImpl(repository: HardCopiesRepository[IO]) extends HardCopiesService[IO] {
override def get(ipfsCid: IpfsCid): IO[Int] = {
repository.get(ipfsCid)
diff --git a/federation-controller-backend/src/main/scala/acab/devcon0/domain/service/ipfscid/SoftCopiesService.scala b/federation-controller-backend/src/main/scala/acab/devcon0/domain/service/ipfscid/SoftCopiesService.scala
new file mode 100644
index 0000000..42fe162
--- /dev/null
+++ b/federation-controller-backend/src/main/scala/acab/devcon0/domain/service/ipfscid/SoftCopiesService.scala
@@ -0,0 +1,9 @@
+package acab.devcon0.domain.service.ipfscid
+
+import acab.devcon0.trile.domain.dtos.aliases.IpfsCid
+
+trait SoftCopiesService[F[_]] {
+ def get(ipfsCid: IpfsCid): F[Int]
+ def set(ipfsCid: IpfsCid, count: Int): F[Unit]
+ def delete(ipfsCid: IpfsCid): F[Unit]
+}
diff --git a/federation-controller-backend/src/main/scala/acab/devcon0/domain/service/ipfscid/SoftCopiesServiceImpl.scala b/federation-controller-backend/src/main/scala/acab/devcon0/domain/service/ipfscid/SoftCopiesServiceImpl.scala
new file mode 100644
index 0000000..d0f7ad6
--- /dev/null
+++ b/federation-controller-backend/src/main/scala/acab/devcon0/domain/service/ipfscid/SoftCopiesServiceImpl.scala
@@ -0,0 +1,20 @@
+package acab.devcon0.domain.service.ipfscid
+import acab.devcon0.domain.ports.output.repository.ipfscid.SoftCopiesRepository
+import acab.devcon0.domain.service.ipfscid
+import acab.devcon0.trile.domain.dtos.aliases.IpfsCid
+import cats.effect.IO
+
+class SoftCopiesServiceImpl(repository: SoftCopiesRepository[IO]) extends SoftCopiesService[IO] {
+
+ override def get(ipfsCid: IpfsCid): IO[Int] = {
+ repository.get(ipfsCid)
+ }
+
+ override def set(ipfsCid: IpfsCid, count: Int): IO[Unit] = {
+ repository.set(ipfsCid, count)
+ }
+
+ override def delete(ipfsCid: IpfsCid): IO[Unit] = {
+ repository.delete(ipfsCid)
+ }
+}
diff --git a/federation-controller-backend/src/main/scala/acab/devcon0/input/redispubsub/FederationMemberSharingFolderUpdateListener.scala b/federation-controller-backend/src/main/scala/acab/devcon0/input/redispubsub/FederationMemberSharingFolderUpdateListener.scala
index dfcfd99..4a0fc3f 100644
--- a/federation-controller-backend/src/main/scala/acab/devcon0/input/redispubsub/FederationMemberSharingFolderUpdateListener.scala
+++ b/federation-controller-backend/src/main/scala/acab/devcon0/input/redispubsub/FederationMemberSharingFolderUpdateListener.scala
@@ -65,13 +65,12 @@ class FederationMemberSharingFolderUpdateListener(
isUpdateChangelogProcessed: Boolean
): IO[Unit] = {
(isUpdateChangelogProcessed, federationMemberUpdateChangelogEvent) match {
- case (true, event: SharingFolderSuccessEvent) => {
+ case (true, event: SharingFolderSuccessEvent) =>
val changelogItem = event.federationMemberChangelogItem
val message = FederationMemberChangelogUpdateMessage(federationMemberId, changelogItem)
redisPublisher
.publish(message)
.onError(logError)
- }
case (_, _) =>
syncCommandHandler.handle(SyncAckCommand(federationMemberId)).flattenEvents >>
logger.info(s"Not publishing isUpdateChangelogProcessed=$isUpdateChangelogProcessed") >>
diff --git a/federation-controller-backend/src/main/scala/acab/devcon0/output/client/IpfsClusterClientImpl.scala b/federation-controller-backend/src/main/scala/acab/devcon0/output/client/IpfsClusterClientImpl.scala
index 0ec574f..0279dd9 100644
--- a/federation-controller-backend/src/main/scala/acab/devcon0/output/client/IpfsClusterClientImpl.scala
+++ b/federation-controller-backend/src/main/scala/acab/devcon0/output/client/IpfsClusterClientImpl.scala
@@ -18,9 +18,9 @@ import sttp.model.Uri
class IpfsClusterClientImpl(
apiUrl: String,
backendResource: Resource[IO, SyncBackend]
-) extends IpfsClusterClient {
+) extends IpfsClusterClient[IO] {
- implicit val logger: Logger[IO] = Slf4jLogger.getLogger[IO]
+ private implicit val logger: Logger[IO] = Slf4jLogger.getLogger[IO]
override def getPeers: IO[IpfsClusterPeers] = {
val uri: Uri = uri"$apiUrl/peers"
@@ -38,6 +38,14 @@ class IpfsClusterClientImpl(
.flatMap(response => IpfsClusterCodecs.Decoders.IpfsClusterPeer(response.body.getOrElse("")))
}
+ override def getPinAllocations(ipfsCid: IpfsCid): IO[IpfsClusterPinAllocation] = {
+ val uri: Uri = uri"$apiUrl/allocations/$ipfsCid"
+ backendResource
+ .use(backend => { IO.blocking(basicRequest.get(uri = uri).readTimeout(1.minutes).send(backend)) })
+ .attemptTap(EffectsUtils.attemptTLog)
+ .flatMap(response => IpfsClusterCodecs.Decoders.IpfsClusterPinAllocation(response.body.getOrElse("")))
+ }
+
override def addPin(ipfsCid: IpfsCid): IO[Unit] = {
val uri: Uri = uri"$apiUrl/pins/$ipfsCid"
backendResource
diff --git a/federation-controller-backend/src/main/scala/acab/devcon0/output/repository/ipfscid/CopiesRepositoryImpl.scala b/federation-controller-backend/src/main/scala/acab/devcon0/output/repository/ipfscid/HardCopiesRepository.scala
similarity index 86%
rename from federation-controller-backend/src/main/scala/acab/devcon0/output/repository/ipfscid/CopiesRepositoryImpl.scala
rename to federation-controller-backend/src/main/scala/acab/devcon0/output/repository/ipfscid/HardCopiesRepository.scala
index 1ad5938..6a8ecf3 100644
--- a/federation-controller-backend/src/main/scala/acab/devcon0/output/repository/ipfscid/CopiesRepositoryImpl.scala
+++ b/federation-controller-backend/src/main/scala/acab/devcon0/output/repository/ipfscid/HardCopiesRepository.scala
@@ -1,6 +1,6 @@
package acab.devcon0.output.repository.ipfscid
-import acab.devcon0.domain.ports.output.repository.ipfscid.CopiesRepository
+import acab.devcon0.domain.ports.output.repository.ipfscid.HardCopiesRepository
import acab.devcon0.output.repository.redisutils
import acab.devcon0.trile.domain.dtos.aliases.IpfsCid
import cats.effect.IO
@@ -14,7 +14,7 @@ import org.typelevel.log4cats.slf4j.Slf4jLogger
private object CopiesRepository {
def key(ipfsCid: IpfsCid): String = {
- String.join(":", redisutils.Redis.SetPrefixes.ipfsCidFlatCopies, ipfsCid)
+ String.join(":", redisutils.Redis.SetPrefixes.ipfsCidFlatHardCopies, ipfsCid)
}
object Operations {
@@ -24,11 +24,11 @@ private object CopiesRepository {
}
def increment(ipfsCids: Set[IpfsCid]): RedisCommands[IO, String, String] => IO[Unit] = { redis =>
- ipfsCids.toList.traverse(ipfsCid => redis.incr(key(ipfsCid))).void
+ ipfsCids.toSeq.traverse(ipfsCid => redis.incr(key(ipfsCid))).void
}
def decrement(ipfsCids: Set[IpfsCid]): RedisCommands[IO, String, String] => IO[Unit] = { redis =>
- ipfsCids.toList.traverse { ipfsCid =>
+ ipfsCids.toSeq.traverse { ipfsCid =>
val redisKey: String = key(ipfsCid)
for
currentVal <- redis.get(redisKey).map(_.getOrElse("0").toInt)
@@ -39,8 +39,8 @@ private object CopiesRepository {
}
}
-class CopiesRepositoryImpl(val commandsApi: Resource[IO, RedisCommands[IO, String, String]])
- extends CopiesRepository[IO] {
+class HardCopiesRepositoryImpl(val commandsApi: Resource[IO, RedisCommands[IO, String, String]])
+ extends HardCopiesRepository[IO] {
implicit val logger: Logger[IO] = Slf4jLogger.getLogger[IO]
diff --git a/federation-controller-backend/src/main/scala/acab/devcon0/output/repository/ipfscid/SearchNameRepositoryImpl.scala b/federation-controller-backend/src/main/scala/acab/devcon0/output/repository/ipfscid/SearchNameRepository.scala
similarity index 100%
rename from federation-controller-backend/src/main/scala/acab/devcon0/output/repository/ipfscid/SearchNameRepositoryImpl.scala
rename to federation-controller-backend/src/main/scala/acab/devcon0/output/repository/ipfscid/SearchNameRepository.scala
diff --git a/federation-controller-backend/src/main/scala/acab/devcon0/output/repository/ipfscid/SoftCopiesRepository.scala b/federation-controller-backend/src/main/scala/acab/devcon0/output/repository/ipfscid/SoftCopiesRepository.scala
new file mode 100644
index 0000000..58fc55c
--- /dev/null
+++ b/federation-controller-backend/src/main/scala/acab/devcon0/output/repository/ipfscid/SoftCopiesRepository.scala
@@ -0,0 +1,80 @@
+package acab.devcon0.output.repository.ipfscid
+
+import acab.devcon0.domain.ports.output.repository.ipfscid.SoftCopiesRepository
+import acab.devcon0.output.repository.redisutils
+import acab.devcon0.trile.domain.dtos.aliases.IpfsCid
+import acab.devcon0.trile.utils.EffectsUtils
+import cats.effect.IO
+import cats.effect.kernel.Resource
+import cats.syntax.all._
+import dev.profunktor.redis4cats._
+import dev.profunktor.redis4cats.tx.TxStore
+import org.typelevel.log4cats.Logger
+import org.typelevel.log4cats.slf4j.Slf4jLogger
+
+private object SoftCopiesRepository {
+
+ def key(ipfsCid: IpfsCid): String = {
+ String.join(":", redisutils.Redis.SetPrefixes.ipfsCidFlatSoftCopies, ipfsCid)
+ }
+
+ object Operations {
+
+ def get(ipfsCid: IpfsCid): RedisCommands[IO, String, String] => IO[Int] = { redis =>
+ redis.get(key(ipfsCid)).map(_.getOrElse("0").toInt)
+ }
+
+ def set(ipfsCid: IpfsCid, count: Int): RedisCommands[IO, String, String] => IO[Unit] = { redis =>
+ redis.set(key(ipfsCid), count.toString).void
+ }
+
+ def delete(ipfsCid: IpfsCid): RedisCommands[IO, String, String] => IO[Unit] = { redis =>
+ redis.del(key(ipfsCid)).void
+ }
+ }
+}
+
+class SoftCopiesRepositoryImpl(val commandsApi: Resource[IO, RedisCommands[IO, String, String]])
+ extends SoftCopiesRepository[IO] {
+
+ private implicit val logger: Logger[IO] = Slf4jLogger.getLogger[IO]
+
+ override def get(ipfsCid: IpfsCid): IO[Int] = {
+ commandsApi
+ .use(SoftCopiesRepository.Operations.get(ipfsCid)(_))
+ }
+
+ override def set(ipfsCid: IpfsCid, count: Int): IO[Unit] = {
+ commandsApi
+ .use(setInner(ipfsCid, count, _))
+ }
+
+ override def delete(ipfsCid: IpfsCid): IO[Unit] = {
+ commandsApi
+ .use(deleteInner(ipfsCid, _))
+ }
+
+ private def setInner(ipfsCid: IpfsCid, count: Int, redis: RedisCommands[IO, String, String]): IO[Unit] = {
+ val operations: TxStore[IO, String, String] => List[IO[Unit]] = { _ =>
+ List(
+ SoftCopiesRepository.Operations.set(ipfsCid, count)(redis)
+ )
+ }
+ redis
+ .transact(operations)
+ .void
+ .attemptTap(EffectsUtils.attemptTLog)
+ }
+
+ private def deleteInner(ipfsCid: IpfsCid, redis: RedisCommands[IO, String, String]): IO[Unit] = {
+ val operations: TxStore[IO, String, String] => List[IO[Unit]] = { _ =>
+ List(
+ SoftCopiesRepository.Operations.delete(ipfsCid)(redis)
+ )
+ }
+ redis
+ .transact(operations)
+ .void
+ .attemptTap(EffectsUtils.attemptTLog)
+ }
+}
diff --git a/federation-controller-backend/src/main/scala/acab/devcon0/output/repository/redisutils/Redis.scala b/federation-controller-backend/src/main/scala/acab/devcon0/output/repository/redisutils/Redis.scala
index d6edaba..2b977d0 100644
--- a/federation-controller-backend/src/main/scala/acab/devcon0/output/repository/redisutils/Redis.scala
+++ b/federation-controller-backend/src/main/scala/acab/devcon0/output/repository/redisutils/Redis.scala
@@ -4,11 +4,12 @@ import dev.profunktor.redis4cats.data.RedisChannel
object Redis {
object SetPrefixes {
- val ipfsCidFlat: String = "IPFS:CID:FLAT"
- val ipfsCidFlatIndex: String = "IPFS:CID:FLAT/INDEX"
- val ipfsCidFlatCopies: String = "IPFS:CID:FLAT/COPIES"
- val ipfsCidNodeTree: String = "IPFS:CID:NODE_TREE"
- val ipfsCidFileTree: String = "IPFS:CID:FILE_TREE"
+ val ipfsCidFlat: String = "IPFS:CID:FLAT"
+ val ipfsCidFlatIndex: String = "IPFS:CID:FLAT/INDEX"
+ val ipfsCidFlatHardCopies: String = "IPFS:CID:FLAT/HARD_COPIES"
+ val ipfsCidFlatSoftCopies: String = "IPFS:CID:FLAT/SOFT_COPIES"
+ val ipfsCidNodeTree: String = "IPFS:CID:NODE_TREE"
+ val ipfsCidFileTree: String = "IPFS:CID:FILE_TREE"
val federationMemberInformation: String = "FEDERATION:MEMBER:INFORMATION"
val federationMemberInformationIndex: String = "FEDERATION:MEMBER:INFORMATION/INDEX"
--
GitLab