Release v1.2.0

src/main/protobuf/grpc-api.proto

@@ -357,6 +357,7 @@ message UserSession {
     int64 expireTime = 3;
     bool isAdmin = 4;
     int64 subscribeEndTime = 5;
+    bool subscribed = 6;
 }
 
 // Define ArticleUserMarking
@@ -532,6 +533,24 @@ message SearchTerms {
     repeated string terms = 1;
 }
 
+// Define ImportSourcesTask
+
+message ImportSourcesTask {
+    string id = 1;
+    string userID = 2;
+    int64 createdAt = 3;
+    int32 totalSources = 4;
+    int32 failedSources = 5;
+    int32 successfulSources = 6;
+}
+
+// Define ImportFailedSource
+
+message ImportFailedSource {
+    string xmlUrl = 1;
+    string error = 2;
+}
+
 // Define me.binwang.rss.service.ArticleService
 
 message GetArticlesBySourceRequest {
@@ -850,6 +869,18 @@ message ImportFromOPMLRequest {
     string inputStream = 2;
 }
 
+message GetImportOPMLTaskRequest {
+    string token = 1;
+}
+
+message GetImportOPMLFailedSourcesRequest {
+    string token = 1;
+}
+
+message DeleteOPMLImportTasksRequest {
+    string token = 1;
+}
+
 message ExportOPMLRequest {
     string token = 1;
 }
@@ -892,9 +923,19 @@ message DeleteFolderRequest {
     string folderID = 2;
 }
 message ImportFromOPMLResponse {
-    int32 result = 1;
+    ImportSourcesTask result = 1;
 }
 
+message GetImportOPMLTaskResponse {
+    ImportSourcesTask result = 1;
+}
+
+message GetImportOPMLFailedSourcesResponse {
+    ImportFailedSource result = 1;
+}
+
+message DeleteOPMLImportTasksResponse {}
+
 message ExportOPMLResponse {
     string result = 1;
 }
@@ -927,6 +968,9 @@ message DeleteFolderResponse {}
 
 service FolderAPI {
     rpc ImportFromOPML (ImportFromOPMLRequest) returns (ImportFromOPMLResponse);
+    rpc GetImportOPMLTask (GetImportOPMLTaskRequest) returns (GetImportOPMLTaskResponse);
+    rpc GetImportOPMLFailedSources (GetImportOPMLFailedSourcesRequest) returns (stream GetImportOPMLFailedSourcesResponse);
+    rpc DeleteOPMLImportTasks (DeleteOPMLImportTasksRequest) returns (DeleteOPMLImportTasksResponse);
     rpc ExportOPML (ExportOPMLRequest) returns (ExportOPMLResponse);
     rpc GetMyFolders (GetMyFoldersRequest) returns (stream GetMyFoldersResponse);
     rpc GetFolderByID (GetFolderByIDRequest) returns (GetFolderByIDResponse);

src/main/resources/application.conf

@@ -24,7 +24,7 @@ redis {
 }
 
 api {
-    version = "1.1.0"
+    version = "1.2.0"
 }
 
 grpc {
@@ -92,13 +92,17 @@ metrics {
 }
 
 article-embedding {
+    enabled = true
+
+    # only needed if enabled is true
     update-interval-millis = 60000 # 1 minute
     cleanup-interval-millis = 86400000 # 1 day
     timeout-millis = 7200000 # 2 hours
     batch-size = 10
     parallelism = 2
 }
 
+# only needed if article-embedding.enabled is true
 ai-server {
     host = "127.0.0.1"
     port = 50001
@@ -113,4 +117,15 @@ search {
 open-ai {
     apiKey = ""
     model = "gpt-3.5-turbo"
+}
+
+image-proxy {
+    host = "http-proxy.rssbrain.com"
+}
+
+import.limit {
+    free-trail-folders = 100
+    free-trail-sources = 200
+    paid-user-folders = 1000
+    paid-user-sources = 2000
 }

src/main/resources/reference.conf

@@ -0,0 +1,8 @@
+
+image-proxy {
+    host = ""
+}
+
+payment {
+    enabled = false
+}

src/main/resources/webview/static/css/main.css

@@ -55,6 +55,13 @@ figcaption {
   display: none !important;
 }
 
+.isDisabled {
+  color: currentColor;
+  cursor: not-allowed;
+  opacity: 0.5;
+  text-decoration: none;
+}
+
 
 /* login page */
 
@@ -765,6 +772,31 @@ progress {
 .import-feed-hint {
   align-self: start;
   color: var(--pico-muted-color);
+  margin-top: 10px;
+}
+
+.opml-import-task-info {
+  font-size: 20px;
+  font-weight: lighter;
+  margin-top: 20px;
+  display: flex;
+  flex-direction: column;
+  gap: 6px;
+}
+
+.opml-import-progress {
+  height: 20px;
+}
+
+.import-err-msg {
+  max-height: 300px;
+  display: flex;
+  width: 100%;
+  overflow-y: scroll;
+  overflow-x: scroll;
+  text-wrap: wrap;
+  flex-direction: column;
+  color: var(--pico-code-color);
 }
 
 #preview-feeds {

src/main/scala/me/binwang/rss/cmd/BaseServer.scala

@@ -32,6 +32,7 @@ class BaseServer(
     val paymentCustomerDao: PaymentCustomerDao,
     val moreLikeThisMappingDao: MoreLikeThisMappingDao,
     val articleEmbeddingTaskDao: ArticleEmbeddingTaskDao,
+    val importSourcesTaskDao: ImportSourcesTaskDao,
     val mailSender: SendGridMailSender,
     val crawler: Crawler,
     val fetcher: BackgroundFetcher,
@@ -77,6 +78,7 @@ object BaseServer {
         implicit val paymentCustomerDao: PaymentCustomerSqlDao = new PaymentCustomerSqlDao()
         implicit val moreLikeThisMappingDao: MoreLikeThisMappingSqlDao = new MoreLikeThisMappingSqlDao()
         implicit val articleEmbeddingTaskDao: ArticleEmbeddingTaskSqlDao = new ArticleEmbeddingTaskSqlDao()
+        implicit val importSourcesTaskDao: ImportSourcesTaskSqlDao = new ImportSourcesTaskSqlDao()
 
         val mailSender = new SendGridMailSender()
 
@@ -98,6 +100,7 @@ object BaseServer {
             paymentCustomerDao.createTable() >>
             moreLikeThisMappingDao.createTable() >>
             articleEmbeddingTaskDao.createTable() >>
+            importSourcesTaskDao.createTable() >>
             BackgroundFetcher(crawler, sourceDao, updater, config.getInt("fetcher.batchSize")).map { fetcher =>
               new BaseServer(
                 articleDao,
@@ -115,6 +118,7 @@ object BaseServer {
                 paymentCustomerDao,
                 moreLikeThisMappingDao,
                 articleEmbeddingTaskDao,
+                importSourcesTaskDao,
                 mailSender,
                 crawler,
                 fetcher,

src/main/scala/me/binwang/rss/cmd/FetchServer.scala

@@ -1,7 +1,7 @@
 package me.binwang.rss.cmd
 
 import cats.effect.unsafe.IORuntime
-import cats.effect.{ExitCode, IO, IOApp}
+import cats.effect.{ExitCode, IO, IOApp, Resource}
 import com.typesafe.config.ConfigFactory
 import me.binwang.rss.fetch.fetcher.ArticleEmbeddingWorker
 import me.binwang.rss.metric.MetricServer
@@ -32,19 +32,27 @@ object FetchServer extends IOApp {
   override def run(args: List[String]): IO[ExitCode] = {
     MetricServer()
       .flatMap { _ => BaseServer()}
-      .flatMap { baseServer => ArticleEmbeddingWorker(
-        baseServer.articleEmbeddingTaskDao,
-        baseServer.articleSearchDao,
-        config.getLong("article-embedding.update-interval-millis"),
-        config.getLong("article-embedding.cleanup-interval-millis"),
-        config.getLong("article-embedding.timeout-millis"),
-        config.getInt("article-embedding.batch-size"),
-        config.getInt("article-embedding.parallelism"),
-        config.getString("ai-server.host"),
-        config.getInt("ai-server.port"))
-        .evalMap { articleEmbeddingWorker =>
-          baseServer.fetcher.run() &> articleEmbeddingWorker.run()
-        }}
+      .flatMap { baseServer =>
+        val workerOpt: Resource[IO, Option[ArticleEmbeddingWorker]] =
+          if (!config.getBoolean("article-embedding.enabled")) {
+            Resource.pure(None)
+          } else {
+            ArticleEmbeddingWorker(
+              baseServer.articleEmbeddingTaskDao,
+              baseServer.articleSearchDao,
+              config.getLong("article-embedding.update-interval-millis"),
+              config.getLong("article-embedding.cleanup-interval-millis"),
+              config.getLong("article-embedding.timeout-millis"),
+              config.getInt("article-embedding.batch-size"),
+              config.getInt("article-embedding.parallelism"),
+              config.getString("ai-server.host"),
+              config.getInt("ai-server.port")
+            ).map(Some(_))
+          }
+        workerOpt.evalMap { articleEmbeddingWorker =>
+          baseServer.fetcher.run() &> articleEmbeddingWorker.map(_.run()).getOrElse(IO.unit)
+        }
+      }
       .useForever
   }
 

src/main/scala/me/binwang/rss/cmd/Services.scala

@@ -4,12 +4,16 @@ import cats.effect.{IO, Resource}
 import cats.implicits._
 import com.typesafe.config.ConfigFactory
 import me.binwang.rss.llm.OpenAILLM
+import me.binwang.rss.model.ImportLimit
 import me.binwang.rss.service._
 import me.binwang.rss.sourcefinder.{HtmlSourceFinder, MultiSourceFinder, RegexSourceFinder}
 import me.binwang.rss.util.Throttler
 import org.typelevel.log4cats.LoggerFactory
 import org.typelevel.log4cats.slf4j.Slf4jFactory
 
+import scala.util.Try
+
+
 case class Services (
     articleService: ArticleService,
     folderService: FolderService,
@@ -35,6 +39,13 @@ object Services {
     val authorizer: Authorizer = new Authorizer(throttler, baseServer.userSessionDao, baseServer.folderDao)
     val llm = new OpenAILLM(baseServer.sttpBackend)
 
+    val importLimit = ImportLimit(
+      paidFolderCount = Try(config.getInt("import.limit.paid-user-folders")).toOption,
+      paidSourceCount = Try(config.getInt("import.limit.paid-user-sources")).toOption,
+      freeFolderCount = Try(config.getInt("import.limit.free-trail-folders")).toOption,
+      freeSourceCount = Try(config.getInt("import.limit.free-trail-sources")).toOption,
+    )
+
     Resource.eval(Seq(
       RegexSourceFinder("rssbrain-regex-rules.json"),
       RegexSourceFinder("rsshub-regex-rules.json"),
@@ -45,7 +56,7 @@ object Services {
         new ArticleService(baseServer.articleDao, baseServer.articleContentDao, baseServer.articleUserMarkingDao,
           baseServer.articleSearchDao, llm, authorizer),
         new FolderService(baseServer.folderDao, baseServer.folderSourceDao,
-          baseServer.sourceDao, authorizer),
+          baseServer.sourceDao, baseServer.importSourcesTaskDao, authorizer, importLimit),
         new SourceService(baseServer.sourceDao, baseServer.folderSourceDao, baseServer.folderDao, baseServer.fetcher,
           authorizer, sourceFinder),
         new UserService(baseServer.userDao, baseServer.userSessionDao, baseServer.userDeleteCodeDao,

src/main/scala/me/binwang/rss/dao/ImportSourcesTaskDao.scala

@@ -0,0 +1,17 @@
+package me.binwang.rss.dao
+
+import cats.effect.IO
+import me.binwang.rss.model.ID.ID
+import me.binwang.rss.model.{ImportFailedSource, ImportSourcesTask, ImportSourcesTaskMapping}
+
+import java.time.ZonedDateTime
+
+trait ImportSourcesTaskDao {
+
+  def insert(task: ImportSourcesTask, sourceMappings: Seq[ImportSourcesTaskMapping]): IO[Unit]
+  def deleteByUser(userID: String): IO[Unit]
+  def getByUserWithUpdatedStats(userID: String, now: ZonedDateTime): IO[Option[ImportSourcesTask]]
+  def updateFailedMessage(taskID: String, sourceIDs: Seq[ID], msg: String): IO[Boolean]
+  def getFailedSources(taskID: String): fs2.Stream[IO, ImportFailedSource]
+
+}

src/main/scala/me/binwang/rss/dao/UserSessionDao.scala

@@ -11,6 +11,6 @@ trait UserSessionDao {
   def get(token: String): IO[Option[UserSession]]
   def delete(token: String): IO[Boolean]
   def deleteByUser(userID: String): IO[Long]
-  def updateSubscribeEndTime(userID: String, subscribeEndTime: ZonedDateTime): IO[Long]
+  def updateSubscription(userID: String, subscribeEndTime: ZonedDateTime, subscribed: Boolean): IO[Long]
 
 }

src/main/scala/me/binwang/rss/dao/hybrid/ArticleHybridDao.scala

@@ -73,8 +73,9 @@ class ArticleHybridDao(val articleSqlDao: ArticleSqlDao, val articleElasticDao:
 
     val insert = for {
       _ <- logger.debug(s"Article changed for ${article.id}, will insert to db")
-      _ <- articleSqlDao.insertOrUpdate(article)
-      _ <- articleElasticDao.insertOrUpdate(article)
+      inserted <- articleSqlDao.insertOrUpdate(article)
+      _ <- if (!inserted) IO.unit else articleElasticDao.insertOrUpdate(article)
+      _ <- if (inserted) IO.unit else MetricReporter.countArticleNotUpdated()
       _ <- MetricReporter.countUpdateArticle(false)
     } yield true